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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
package protocol
import (
v1 "github.com/golang/protobuf/proto" //nolint:staticcheck
jsoniter "github.com/json-iterator/go"
"github.com/spiral/errors"
"github.com/spiral/roadrunner/v2/pkg/payload"
"github.com/spiral/roadrunner/v2/plugins/logger"
"github.com/spiral/roadrunner/v2/plugins/temporal/protocol/internal"
"google.golang.org/protobuf/proto"
)
type (
// ProtoCodec uses protobuf to exchange messages with underlying workers.
ProtoCodec struct {
}
)
// NewProtoCodec creates new Proto communication codec.
func NewProtoCodec() Codec {
return &ProtoCodec{}
}
// WithLogger creates new codes instance with attached logger.
func (c *ProtoCodec) WithLogger(logger logger.Logger) Codec {
return &ProtoCodec{}
}
// GetName returns codec name.
func (c *ProtoCodec) GetName() string {
return "protobuf"
}
// Execute exchanges commands with worker.
func (c *ProtoCodec) Execute(e Endpoint, ctx Context, msg ...Message) ([]Message, error) {
if len(msg) == 0 {
return nil, nil
}
var request = &internal.Frame{}
var response = &internal.Frame{}
var result = make([]Message, 0, 5)
var err error
for _, m := range msg {
frame, err := c.packMessage(m)
if err != nil {
return nil, err
}
request.Messages = append(request.Messages, frame)
}
p := payload.Payload{}
// context is always in json format
if ctx.IsEmpty() {
p.Context = []byte("null")
}
p.Context, err = jsoniter.Marshal(ctx)
if err != nil {
return nil, errors.E(errors.Op("encodeContext"), err)
}
p.Body, err = proto.Marshal(v1.MessageV2(request))
if err != nil {
return nil, errors.E(errors.Op("encodePayload"), err)
}
out, err := e.Exec(p)
if err != nil {
return nil, errors.E(errors.Op("execute"), err)
}
if len(out.Body) == 0 {
// worker inactive or closed
return nil, nil
}
err = proto.Unmarshal(out.Body, v1.MessageV2(response))
if err != nil {
return nil, errors.E(errors.Op("parseResponse"), err)
}
for _, f := range response.Messages {
msg, err := c.parseMessage(f)
if err != nil {
return nil, err
}
result = append(result, msg)
}
return result, nil
}
func (c *ProtoCodec) packMessage(msg Message) (*internal.Message, error) {
var err error
frame := &internal.Message{
Id: msg.ID,
Payloads: msg.Payloads,
Failure: msg.Failure,
}
if msg.Command != nil {
frame.Command, err = commandName(msg.Command)
if err != nil {
return nil, err
}
frame.Options, err = jsoniter.Marshal(msg.Command)
if err != nil {
return nil, err
}
}
return frame, nil
}
func (c *ProtoCodec) parseMessage(frame *internal.Message) (Message, error) {
const op = errors.Op("proto_codec_parse_message")
var err error
msg := Message{
ID: frame.Id,
Payloads: frame.Payloads,
Failure: frame.Failure,
}
if frame.Command != "" {
msg.Command, err = initCommand(frame.Command)
if err != nil {
return Message{}, errors.E(op, err)
}
err = jsoniter.Unmarshal(frame.Options, &msg.Command)
if err != nil {
return Message{}, errors.E(op, err)
}
}
return msg, nil
}
|
