client support for rpc cancel (#249)

This commit is contained in:
Mike Sawka 2024-08-19 15:44:30 -07:00 committed by GitHub
parent 00958b8fed
commit 534fcc5d0a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 52 additions and 22 deletions

View File

@ -51,7 +51,11 @@ async function* rpcResponseGenerator(
return; return;
} }
const shouldTerminate = yield msg.data; const shouldTerminate = yield msg.data;
if (shouldTerminate || !msg.cont) { if (shouldTerminate) {
sendRpcCancel(reqid);
return;
}
if (!msg.cont) {
return; return;
} }
} }
@ -65,6 +69,12 @@ async function* rpcResponseGenerator(
} }
} }
function sendRpcCancel(reqid: string) {
const rpcMsg: RpcMessage = { reqid: reqid, cancel: true };
const wsMsg: WSRpcCommand = { wscommand: "rpc", message: rpcMsg };
globalWS.pushMessage(wsMsg);
}
function sendRpcCommand(msg: RpcMessage): AsyncGenerator<RpcMessage, void, boolean> { function sendRpcCommand(msg: RpcMessage): AsyncGenerator<RpcMessage, void, boolean> {
const wsMsg: WSRpcCommand = { wscommand: "rpc", message: msg }; const wsMsg: WSRpcCommand = { wscommand: "rpc", message: msg };
globalWS.pushMessage(wsMsg); globalWS.pushMessage(wsMsg);

View File

@ -79,6 +79,13 @@ func generateTSMethodTypes(method reflect.Method, tsTypesMap map[reflect.Type]st
} }
func getTSFieldName(field reflect.StructField) string { func getTSFieldName(field reflect.StructField) string {
tsFieldTag := field.Tag.Get("tsfield")
if tsFieldTag != "" {
if tsFieldTag == "-" {
return ""
}
return tsFieldTag
}
jsonTag := field.Tag.Get("json") jsonTag := field.Tag.Get("json")
if jsonTag != "" { if jsonTag != "" {
parts := strings.Split(jsonTag, ",") parts := strings.Split(jsonTag, ",")

View File

@ -57,27 +57,30 @@ func sendRpcRequestResponseStreamHelper[T any](w *wshutil.WshRpc, command string
if err != nil { if err != nil {
rtnErr(respChan, err) rtnErr(respChan, err)
return respChan return respChan
} else {
go func() {
defer close(respChan)
for {
if reqHandler.ResponseDone() {
break
}
resp, err := reqHandler.NextResponse()
if err != nil {
respChan <- wshrpc.RespOrErrorUnion[T]{Error: err}
break
}
var respData T
err = utilfn.ReUnmarshal(&respData, resp)
if err != nil {
respChan <- wshrpc.RespOrErrorUnion[T]{Error: err}
break
}
respChan <- wshrpc.RespOrErrorUnion[T]{Response: respData}
}
}()
} }
opts.StreamCancelFn = func() {
// TODO coordinate the cancel with the for loop below
reqHandler.SendCancel()
}
go func() {
defer close(respChan)
for {
if reqHandler.ResponseDone() {
break
}
resp, err := reqHandler.NextResponse()
if err != nil {
respChan <- wshrpc.RespOrErrorUnion[T]{Error: err}
break
}
var respData T
err = utilfn.ReUnmarshal(&respData, resp)
if err != nil {
respChan <- wshrpc.RespOrErrorUnion[T]{Error: err}
break
}
respChan <- wshrpc.RespOrErrorUnion[T]{Response: respData}
}
}()
return respChan return respChan
} }

View File

@ -105,6 +105,8 @@ type RpcOpts struct {
Timeout int `json:"timeout,omitempty"` Timeout int `json:"timeout,omitempty"`
NoResponse bool `json:"noresponse,omitempty"` NoResponse bool `json:"noresponse,omitempty"`
Route string `json:"route,omitempty"` Route string `json:"route,omitempty"`
StreamCancelFn func() `json:"-"` // this is an *output* parameter, set by the handler
} }
type RpcContext struct { type RpcContext struct {

View File

@ -76,6 +76,14 @@ func GetRpcSourceFromContext(ctx context.Context) string {
return rtn.(*RpcResponseHandler).GetSource() return rtn.(*RpcResponseHandler).GetSource()
} }
func GetIsCanceledFromContext(ctx context.Context) bool {
rtn := ctx.Value(wshRpcRespHandlerContextKey{})
if rtn == nil {
return false
}
return rtn.(*RpcResponseHandler).IsCanceled()
}
func GetRpcResponseHandlerFromContext(ctx context.Context) *RpcResponseHandler { func GetRpcResponseHandlerFromContext(ctx context.Context) *RpcResponseHandler {
rtn := ctx.Value(wshRpcRespHandlerContextKey{}) rtn := ctx.Value(wshRpcRespHandlerContextKey{})
if rtn == nil { if rtn == nil {