Method MessageLoop

server/hdl_grpc.go:38–74

Equivalent of starting a new session and a read loop in one.

(stream pbx.Node_MessageLoopServer)

// Equivalent of starting a new session and a read loop in one.
func (*grpcNodeServer) MessageLoop(stream pbx.Node_MessageLoopServer) error {
	sess, count := globals.sessionStore.NewSession(stream, "")
	if p, ok := peer.FromContext(stream.Context()); ok {
		sess.remoteAddr = p.Addr.String()
	}
	logs.Info.Println("grpc: session started", sess.sid, sess.remoteAddr, count)

	defer func() {
		sess.closeGrpc()
		sess.cleanUp(false)
	}()

	go sess.writeGrpcLoop()

	for {
		in, err := stream.Recv()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			logs.Err.Println("grpc: recv", sess.sid, err)
			return err
		}
		logs.Info.Println("grpc in:", truncateStringIfTooLong(in.String()), sess.sid)
		statsInc("IncomingMessagesGrpcTotal", 1)
		sess.dispatch(pbCliDeserialize(in))

		sess.lock.Lock()
		if sess.grpcnode == nil {
			sess.lock.Unlock()
			break
		}
		sess.lock.Unlock()
	}

	return nil
}

func (sess *Session) sendMessageGrpc(msg any) bool {
	if len(sess.send) > sendQueueLimit {