Skip to content

Commit 2fc64f8

Browse files
committed
feat(tars2go): Support rpc asynchronous callback call
1 parent fe8a4e9 commit 2fc64f8

File tree

3 files changed

+303
-40
lines changed

3 files changed

+303
-40
lines changed

tars/message.go

+1
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,7 @@ type Message struct {
3838
isHash bool
3939
Async bool
4040
Callback model.Callback
41+
RespCh chan *requestf.ResponsePacket
4142
}
4243

4344
// Init define the beginTime

tars/servant.go

+8-13
Original file line numberDiff line numberDiff line change
@@ -129,9 +129,7 @@ func (s *ServantProxy) TarsInvoke(ctx context.Context, cType byte,
129129

130130
msg := buildMessage(ctx, cType, sFuncName, buf, status, reqContext, resp, s)
131131
timeout := time.Duration(s.syncTimeout) * time.Millisecond
132-
err := s.invokeFilters(ctx, msg, timeout)
133-
134-
if err != nil {
132+
if err := s.invokeFilters(ctx, msg, timeout); err != nil {
135133
return err
136134
}
137135
*resp = *msg.Resp
@@ -238,8 +236,8 @@ func (s *ServantProxy) doInvoke(ctx context.Context, msg *Message, timeout time.
238236
}
239237

240238
atomic.AddInt32(&s.queueLen, 1)
241-
readCh := make(chan *requestf.ResponsePacket)
242-
adp.resp.Store(msg.Req.IRequestId, readCh)
239+
msg.RespCh = make(chan *requestf.ResponsePacket)
240+
adp.resp.Store(msg.Req.IRequestId, msg.RespCh)
243241
var releaseFunc = func() {
244242
CheckPanic()
245243
atomic.AddInt32(&s.queueLen, -1)
@@ -265,7 +263,7 @@ func (s *ServantProxy) doInvoke(ctx context.Context, msg *Message, timeout time.
265263
if msg.Async {
266264
go func() {
267265
defer releaseFunc()
268-
err := s.waitInvoke(msg, adp, timeout, needCheck)
266+
err := s.waitResp(msg, timeout, needCheck)
269267
s.manager.postInvoke()
270268
msg.End()
271269
s.reportStat(msg, err)
@@ -280,21 +278,18 @@ func (s *ServantProxy) doInvoke(ctx context.Context, msg *Message, timeout time.
280278
return nil
281279
}
282280

283-
return s.waitInvoke(msg, adp, timeout, needCheck)
281+
return s.waitResp(msg, timeout, needCheck)
284282
}
285283

286-
func (s *ServantProxy) waitInvoke(msg *Message, adp *AdapterProxy, timeout time.Duration, needCheck bool) error {
287-
ch, _ := adp.resp.Load(msg.Req.IRequestId)
288-
readCh := ch.(chan *requestf.ResponsePacket)
289-
284+
func (s *ServantProxy) waitResp(msg *Message, timeout time.Duration, needCheck bool) error {
285+
adp := msg.Adp
290286
select {
291287
case <-rtimer.After(timeout):
292288
msg.Status = basef.TARSINVOKETIMEOUT
293289
adp.failAdd()
294-
msg.End()
295290
return fmt.Errorf("request timeout, begin time:%d, cost:%d, obj:%s, func:%s, addr:(%s:%d), reqid:%d",
296291
msg.BeginTime, msg.Cost(), msg.Req.SServantName, msg.Req.SFuncName, adp.point.Host, adp.point.Port, msg.Req.IRequestId)
297-
case msg.Resp = <-readCh:
292+
case msg.Resp = <-msg.RespCh:
298293
if needCheck {
299294
go func() {
300295
adp.reset()

0 commit comments

Comments
 (0)