Method runServer

pkg/wshutil/wshrpc.go:381–447

()

}

func (w *WshRpc) runServer() {
	defer func() {
		panichandler.PanicHandler("wshrpc.runServer", recover())
		close(w.OutputCh)
		w.setServerDone()
	}()
outer:
	for {
		var inputVal baseds.RpcInputChType
		var inputChMore bool
		var resIdTimeout string

		select {
		case inputVal, inputChMore = <-w.InputCh:
			if !inputChMore {
				break outer
			}
			if w.Debug {
				log.Printf("[%s] received message: %s\n", w.DebugName, string(inputVal.MsgBytes))
			}
		case resIdTimeout = <-w.CtxDoneCh:
			if w.Debug {
				log.Printf("[%s] received request timeout: %s\n", w.DebugName, resIdTimeout)
			}
			w.unregisterRpc(resIdTimeout, fmt.Errorf("EC-TIME: timeout waiting for response"))
			continue
		}

		var msg RpcMessage
		err := json.Unmarshal(inputVal.MsgBytes, &msg)
		if err != nil {
			log.Printf("wshrpc received bad message: %v\n", err)
			continue
		}
		if msg.Cancel {
			if msg.ReqId != "" {
				w.cancelRequest(msg.ReqId)
			}
			continue
		}
		if msg.IsRpcRequest() {
			// Handle stream commands synchronously since the broker is designed to be non-blocking.
			// RecvData/RecvAck just enqueue to work queues, so there's no risk of blocking the main loop.
			if msg.Command == wshrpc.Command_StreamData {
				w.handleStreamData(&msg)
				continue
			}
			if msg.Command == wshrpc.Command_StreamDataAck {
				w.handleStreamAck(&msg)
				continue
			}

			ingressLinkId := inputVal.IngressLinkId
			go func() {
				defer func() {
					panichandler.PanicHandler("handleRequest:goroutine", recover())
				}()
				w.handleRequest(&msg, ingressLinkId)