Handle the stream sending in thread safe way

keepalive/keep_alive.go

@@ -44,6 +44,7 @@ func (k *KeepAlive) StreamInterceptor() grpc.StreamServerInterceptor {
 		}
 
 		m := &ioMonitor{
+			sync.Mutex{},
 			sync.Mutex{},
 			stream,
 			time.Now(),

keepalive/monitor.go

@@ -8,13 +8,16 @@ import (
 )
 
 type ioMonitor struct {
-	mu sync.Mutex
+	mu         sync.Mutex
+	streamLock sync.Mutex
 	grpc.ServerStream
 	lastSeen time.Time
 }
 
 func (l *ioMonitor) sendMsg(m interface{}) error {
 	l.updateLastSeen()
+	l.streamLock.Lock()
+	defer l.streamLock.Unlock()
 	return l.ServerStream.SendMsg(m)
 }