Merge pull request #9 from handsomewwu/main

support stream plugin

client/client.go

@@ -11,6 +11,7 @@ import (
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/handler"
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/logger"
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
+	"github.com/open-dingtalk/dingtalk-stream-sdk-go/plugin"
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/utils"
 	"io"
 	"net/http"
@@ -380,6 +381,11 @@ func (cli *StreamClient) RegisterChatBotCallbackRouter(messageHandler chatbot.IC
 	cli.RegisterRouter(utils.SubscriptionTypeKCallback, payload.BotMessageCallbackTopic, chatbot.NewDefaultChatBotFrameHandler(messageHandler).OnEventReceived)
 }
 
+// AI插件的注册函数
+func (cli *StreamClient) RegisterPluginCallbackRouter(messageHandler plugin.IPluginMessageHandler) {
+	cli.RegisterRouter(utils.SubscriptionTypeKCallback, payload.PluginMessageCallbackTopic, plugin.NewDefaultPluginFrameHandler(messageHandler).OnEventReceived)
+}
+
 // 事件类型的注册函数
 func (cli *StreamClient) RegisterEventRouter(topic string, frameHandler handler.IFrameHandler) {
 	cli.RegisterRouter(utils.SubscriptionTypeKEvent, topic, frameHandler)

example/echo_service.go

@@ -0,0 +1,15 @@
+package main
+
+// 插件例子
+type EchoRequest struct {
+	Message string
+}
+type EchoResponse struct {
+	EchoMessage string
+}
+
+func Echo(echoRequest *EchoRequest) *EchoResponse {
+	return &EchoResponse{
+		EchoMessage: "You said: " + echoRequest.Message,
+	}
+}

example/example.go

@@ -9,6 +9,7 @@ import (
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/event"
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/logger"
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
+	"github.com/open-dingtalk/dingtalk-stream-sdk-go/plugin"
 )
 
 /**
@@ -27,6 +28,23 @@ func OnChatBotMessageReceived(ctx context.Context, data *chatbot.BotCallbackData
 	return []byte(""), nil
 }
 
+// 简单的插件处理实现
+func OnPluginMessageReceived(ctx context.Context, message *plugin.PluginMessage) (interface{}, error) {
+	//可以根据message中的PluginId、PluginVersion、AbilityKey路由到具体一个能力
+	if message.AbilityKey == "echo" {
+		echoRequest := &EchoRequest{}
+		//将数据转换成插件的请求参数
+		err := message.ParseRequest(echoRequest)
+		if err != nil {
+			return nil, err
+		}
+		//执行插件
+		echoResponse := Echo(echoRequest)
+		return echoResponse, nil
+	}
+	return nil, nil
+}
+
 // 事件处理
 func OnEventReceived(ctx context.Context, df *payload.DataFrame) (frameResp *payload.DataFrameResponse, err error) {
 	eventHeader := event.NewEventHeaderFromDataFrame(df)
@@ -61,6 +79,8 @@ func main() {
 	cli.RegisterAllEventRouter(OnEventReceived)
 	//注册callback类型的处理函数
 	cli.RegisterChatBotCallbackRouter(OnChatBotMessageReceived)
+	//注册插件的处理函数
+	cli.RegisterPluginCallbackRouter(OnPluginMessageReceived)
 
 	err := cli.Start(context.Background())
 	if err != nil {

payload/utils.go

@@ -24,6 +24,7 @@ const (
 	DataFrameResponseStatusCodeKHandlerNotFound = 404
 
 	BotMessageCallbackTopic    = "/v1.0/im/bot/messages/get"  //机器人消息统一回调topic
+	PluginMessageCallbackTopic = "/v1.0/agi/plugins/callback" //AI插件消息统一回调topic
 )
 
 func GenerateMessageId(prefix string) string {

plugin/model.go

@@ -0,0 +1,31 @@
+package plugin
+
+import (
+	"encoding/json"
+)
+
+type PluginMessage struct {
+	PluginId      string      `json:"pluginId"`
+	PluginVersion string      `json:"pluginVersion"`
+	AbilityKey    string      `json:"abilityKey"`
+	Data          interface{} `json:"data"`
+	RequestId     string      `json:"requestId"`
+}
+
+// 用于将数据转换成插件的请求参数
+func (req *PluginMessage) ParseRequest(pluginRequest interface{}) error {
+	data, err := json.Marshal(req.Data)
+	if err != nil {
+		return err
+	}
+	err = json.Unmarshal(data, pluginRequest)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+type PluginResponse struct {
+	Result    interface{} `json:"result"`
+	RequestId string      `json:"requestId"`
+}

plugin/plugin_handler.go

@@ -0,0 +1,45 @@
+package plugin
+
+import (
+	"context"
+	"encoding/json"
+	"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
+)
+
+type CallbackResponse struct {
+	Response interface{} `json:"response"`
+}
+
+type IPluginMessageHandler func(c context.Context, data *PluginMessage) (interface{}, error)
+
+type DefaultPluginFrameHandler struct {
+	defaultHandler IPluginMessageHandler
+}
+
+func NewDefaultPluginFrameHandler(defaultHandler IPluginMessageHandler) *DefaultPluginFrameHandler {
+	return &DefaultPluginFrameHandler{
+		defaultHandler: defaultHandler,
+	}
+}
+
+func (h *DefaultPluginFrameHandler) OnEventReceived(ctx context.Context, df *payload.DataFrame) (*payload.DataFrameResponse, error) {
+	msgData := &PluginMessage{}
+	err := json.Unmarshal([]byte(df.Data), msgData)
+	if err != nil {
+		return nil, err
+	}
+
+	if h.defaultHandler == nil {
+		return payload.NewDataFrameResponse(payload.DataFrameResponseStatusCodeKHandlerNotFound), nil
+	}
+
+	result, err := h.defaultHandler(ctx, msgData)
+	if err != nil {
+		return nil, err
+	}
+	pluginResponse := &PluginResponse{RequestId: msgData.RequestId, Result: result}
+	callbackResponse := &CallbackResponse{Response: pluginResponse}
+	frameResp := payload.NewSuccessDataFrameResponse()
+	frameResp.SetJson(callbackResponse)
+	return frameResp, nil
+}