Method handleRequestInternal

pkg/wshutil/wshrpc.go:327–379

(req *RpcMessage, ingressLinkId baseds.LinkId, pprofCtx context.Context)

}

func (w *WshRpc) handleRequestInternal(req *RpcMessage, ingressLinkId baseds.LinkId, pprofCtx context.Context) {
	if req.Command == wshrpc.Command_EventRecv {
		w.handleEventRecv(req)
		return
	}

	var respHandler *RpcResponseHandler
	timeoutMs := req.Timeout
	if timeoutMs <= 0 {
		timeoutMs = DefaultTimeoutMs
	}
	ctx, cancelFn := context.WithTimeout(context.Background(), time.Duration(timeoutMs)*time.Millisecond)
	ctx = withWshRpcContext(ctx, w)
	respHandler = &RpcResponseHandler{
		w:               w,
		ctx:             ctx,
		reqId:           req.ReqId,
		command:         req.Command,
		commandData:     req.Data,
		source:          req.Source,
		ingressLinkId:   ingressLinkId,
		done:            &atomic.Bool{},
		canceled:        &atomic.Bool{},
		contextCancelFn: &atomic.Pointer[context.CancelFunc]{},
		rpcCtx:          w.GetRpcContext(),
	}
	respHandler.contextCancelFn.Store(&cancelFn)
	respHandler.ctx = withRespHandler(ctx, respHandler)
	if req.ReqId != "" {
		w.registerResponseHandler(req.ReqId, respHandler)
	}
	isAsync := false
	defer func() {
		panicErr := panichandler.PanicHandler("handleRequest", recover())
		if panicErr != nil {
			respHandler.SendResponseError(panicErr)
		}
		if isAsync {
			go func() {
				defer func() {
					panichandler.PanicHandler("handleRequest:finalize", recover())
				}()
				<-ctx.Done()
				respHandler.Finalize()
			}()
		} else {
			cancelFn()
			respHandler.Finalize()
		}
	}()
	handlerFn := serverImplAdapter(w.ServerImpl)
	isAsync = !handlerFn(respHandler)
}

func (w *WshRpc) runServer() {
	defer func() {