-
Notifications
You must be signed in to change notification settings - Fork 2
/
robot.go
200 lines (162 loc) · 4.16 KB
/
robot.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
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
package robots
import (
"context"
"encoding/json"
"errors"
"github.com/ArtisanCloud/RobotChat/robots/kernel/logger"
"github.com/ArtisanCloud/RobotChat/robots/kernel/model"
queue2 "github.com/ArtisanCloud/RobotChat/robots/kernel/queue"
"sync"
)
type Robot struct {
*model.RobotAttributes
// components
Queue queue2.QueueInterface
Logger *logger.Logger
// Middlewares
ErrorHandler model.HandleError
PreMessageHandlers []model.HandlePreSend
PostMessageHandlers []model.HandlePostReply
// functional
IsWorking bool
Mutex sync.Mutex
ErrorChan chan *model.ErrReply
// webhook
NotifyUrl string
}
func NewRobot(attributes *model.RobotAttributes) (*Robot, error) {
// 返回Robot
return &Robot{
RobotAttributes: attributes,
IsWorking: false,
Mutex: sync.Mutex{},
ErrorChan: make(chan *model.ErrReply),
}, nil
}
func (bot *Robot) SetMessagePreHandler(handles ...model.HandlePreSend) {
bot.PreMessageHandlers = append(bot.PreMessageHandlers, handles...)
}
func (bot *Robot) SetPostMessageHandler(handles ...model.HandlePostReply) {
bot.PostMessageHandlers = append(bot.PostMessageHandlers, handles...)
}
func (bot *Robot) SetErrorHandler(handle model.HandleError) {
bot.ErrorHandler = handle
}
func (bot *Robot) IsAwaken(ctx context.Context) error {
if !bot.Queue.IsConnected(ctx) {
return errors.New("queue is not connected")
}
return nil
}
func (bot *Robot) Start(ctx context.Context) error {
// 检查是否已经唤醒
if bot.IsWorking {
return nil
}
// 加锁
bot.Mutex.Lock()
defer bot.Mutex.Unlock()
//
err := bot.IsAwaken(ctx)
if err != nil {
return err
}
// 启动消费消息的 Goroutine
go bot.Receive(ctx)
// 监听错误通道并处理错误
go func() {
for errReply := range bot.ErrorChan {
// 调用错误处理函数进行处理
if bot.ErrorHandler != nil {
bot.ErrorHandler(errReply)
}
}
}()
// 设置唤醒标志位
bot.IsWorking = true
return nil
}
func (bot *Robot) Stop() {
// 加锁
bot.Mutex.Lock()
defer bot.Mutex.Unlock()
// 关闭 errorChan
close(bot.ErrorChan)
// 设置工作状态为停止
bot.IsWorking = false
}
func (bot *Robot) CreateTextMessage(content interface{}) (*model.Message, error) {
return bot.CreateMessage(model.TextMessage, content)
}
func (bot *Robot) CreateImageMessage(content interface{}) (*model.Message, error) {
return bot.CreateMessage(model.ImageMessage, content)
}
func (bot *Robot) CreateMessage(messageType model.MessageType, content interface{}) (*model.Message, error) {
strContent, err := json.Marshal(content)
if err != nil {
return nil, err
}
message := model.NewMessage(messageType)
message.MessageType = messageType
message.Content = strContent
message.Metadata = model.MetaData{
Robot: bot.RobotAttributes,
Conversation: nil,
}
return message, nil
}
func (bot *Robot) Send(ctx context.Context, message *model.Message) (*model.Job, error) {
job := &model.Job{
Id: model.GenerateId(),
Payload: message,
}
// 将消息传递给中间件处理
for _, middleware := range bot.PreMessageHandlers {
var err error
// 执行中间件处理逻辑
job, err = middleware(ctx, job)
if err != nil {
// 如果中间件返回错误,可以选择处理错误或直接返回
return nil, err
}
}
err := bot.Queue.ProduceMessage(ctx, job)
if err != nil {
return nil, err
}
return job, nil
}
func (bot *Robot) Receive(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
default:
// 从队列中获取消息
job, err := bot.Queue.ConsumeMessage(ctx)
if err != nil {
errReply := &model.ErrReply{
ctx, job, err,
}
// 处理获取消息的错误
bot.ErrorChan <- errReply
// 可以选择进行错误处理或直接返回
continue
}
// 将消息传递给中间件处理
for _, middleware := range bot.PostMessageHandlers {
var err error
// 执行中间件处理逻辑
job, err = middleware(ctx, job)
if err != nil {
errReply := &model.ErrReply{
ctx, job, err,
}
bot.ErrorChan <- errReply
// 如果中间件返回错误,可以选择处理错误或直接返回
continue
}
}
}
}
}