-
Notifications
You must be signed in to change notification settings - Fork 1
/
gate.go
76 lines (66 loc) · 2.24 KB
/
gate.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
package processor
import (
"context"
"git.golaxy.org/core/runtime"
"git.golaxy.org/core/service"
"git.golaxy.org/core/util/generic"
"git.golaxy.org/core/util/types"
"git.golaxy.org/framework/net/gap"
"git.golaxy.org/framework/net/gap/codec"
"git.golaxy.org/framework/plugins/dentq"
"git.golaxy.org/framework/plugins/dserv"
"git.golaxy.org/framework/plugins/gate"
"git.golaxy.org/framework/plugins/log"
"git.golaxy.org/framework/plugins/router"
"git.golaxy.org/framework/util/concurrent"
)
// NewGateProcessor 创建网关RPC处理器,用于C<->G的通信
func NewGateProcessor(mc gap.IMsgCreator) any {
return &_GateProcessor{
encoder: codec.MakeEncoder(),
decoder: codec.MakeDecoder(mc),
}
}
// _GateProcessor 网关RPC处理器,用于C<->G的通信
type _GateProcessor struct {
servCtx service.Context
dist dserv.IDistService
dentq dentq.IDistEntityQuerier
gate gate.IGate
router router.IRouter
encoder codec.Encoder
decoder codec.Decoder
sessionWatcher gate.IWatcher
msgWatcher dserv.IWatcher
}
// Init 初始化
func (p *_GateProcessor) Init(ctx service.Context) {
p.servCtx = ctx
p.dist = dserv.Using(ctx)
p.dentq = dentq.Using(ctx)
p.gate = gate.Using(ctx)
p.router = router.Using(ctx)
p.sessionWatcher = p.gate.Watch(context.Background(), generic.MakeDelegateAction3(p.handleSessionChanged))
p.msgWatcher = p.dist.WatchMsg(context.Background(), generic.MakeDelegateFunc2(p.handleMsg))
log.Debugf(p.servCtx, "rpc processor %q started", types.AnyFullName(*p))
}
// Shut 结束
func (p *_GateProcessor) Shut(ctx service.Context) {
<-p.sessionWatcher.Terminate()
<-p.msgWatcher.Terminate()
log.Debugf(p.servCtx, "rpc processor %q stopped", types.AnyFullName(*p))
}
// Match 是否匹配
func (p *_GateProcessor) Match(ctx service.Context, dst, path string, oneWay bool) bool {
return false
}
// Request 请求
func (p *_GateProcessor) Request(ctx service.Context, dst, path string, args []any) runtime.AsyncRet {
ret := concurrent.MakeRespAsyncRet()
ret.Push(concurrent.MakeRet[any](nil, ErrUndeliverable))
return ret.CastAsyncRet()
}
// Notify 通知
func (p *_GateProcessor) Notify(ctx service.Context, dst, path string, args []any) error {
return ErrUndeliverable
}