Merge pull request #19 from oldduckruirui/main

add demo for ZMTP

Author: rvql

example/zmtp/.gitignore

@@ -0,0 +1 @@
+/pb

example/zmtp/demo.proto

@@ -0,0 +1,24 @@
+syntax = "proto3";
+
+package pb;
+
+option go_package = ".;pb";
+
+service Game {
+  rpc Game(OrderRequest) returns (OrderResponse);
+}
+
+message OrderRequest {
+  string business_id = 1235;
+}
+
+message OrderResponse {
+  string msg = 1235;
+}
+
+message MessageWrapper {
+    oneof message {
+        OrderRequest request = 1;
+        OrderResponse response = 2;
+    }
+}

example/zmtp/wasm.wasm

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"encoding/json"
+
+	"github.com/deepflowio/deepflow-wasm-go-sdk/example/zmtp/pb"
+	"github.com/deepflowio/deepflow-wasm-go-sdk/sdk"
+	sdkpb "github.com/deepflowio/deepflow-wasm-go-sdk/sdk/pb"
+)
+
+//go:generate mkdir -p pb
+//go:generate protoc --go_out=./pb --go-vtproto_out=./pb --go-vtproto_opt=features=unmarshal ./demo.proto
+func main() {
+	sdk.Info("zmtp-plugin loaded")
+	parser := ZrpcParser{}
+	parser.Parser = interface{}(parser).(sdk.Parser)
+	sdk.SetParser(parser)
+}
+
+type ZrpcParser struct {
+	sdk.DefaultParser
+}
+
+func (p ZrpcParser) HookIn() []sdk.HookBitmap {
+	return []sdk.HookBitmap{
+		sdk.HOOK_POINT_CUSTOM_MESSAGE,
+	}
+}
+
+func (p ZrpcParser) CustomMessageHookIn() uint64 {
+	return sdk.CustomMessageHookProtocol(sdk.PROTOCOL_ZMTP, true)
+}
+
+func (p ZrpcParser) onZMTPMessage(payload []byte) sdk.Action {
+	var zmtpMsg sdkpb.ZmtpMessage
+	if err := zmtpMsg.UnmarshalVT(payload); err != nil {
+		return sdk.ActionNext()
+	}
+	var msgWrapper pb.MessageWrapper
+	if err := msgWrapper.UnmarshalVT(zmtpMsg.Payload); err != nil {
+		return sdk.ActionNext()
+	}
+	jsonData, err := json.Marshal(&msgWrapper)
+	if err != nil {
+		jsonData = nil
+	}
+	jsonStr := string(jsonData)
+	return sdk.CustomMessageActionAbortWithResult([]sdk.KeyVal{
+		{
+			Key: "json_payload",
+			Val: jsonStr,
+		},
+	})
+}
+
+func (p ZrpcParser) OnCustomMessage(ctx *sdk.CustomMessageCtx) sdk.Action {
+	if ctx.HookPoint == sdk.ProtocolParse && ctx.TypeCode == uint32(sdk.PROTOCOL_ZMTP) {
+		return p.onZMTPMessage(ctx.Payload)
+	} else {
+		return sdk.ActionNext()
+	}
+}

example/zmtp/zmtp.go

@@ -8,3 +8,10 @@ message NatsMessage {
     string reply_to = 2;
     bytes payload = 3;
 }
+
+message ZmtpMessage {
+    oneof subscription {
+        string match_pattern = 1;
+    }
+    bytes payload = 2;
+}

sdk/WasmPluginApi.proto

@@ -54,6 +54,7 @@ var (
 
 var (
 	PROTOCOL_NATS uint16 = 104
+	PROTOCOL_ZMTP uint16 = 106
 )
 
 var CUSTOM_MESSAGE_HOOK_ALL uint64 = 0xff << 48