You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by mi...@apache.org on 2022/09/15 11:28:49 UTC
[incubator-eventmesh] branch eventmesh-server-go updated: add rocketmq message convert
This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch eventmesh-server-go
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/eventmesh-server-go by this push:
new 56ba18cc add rocketmq message convert
new e5c57fa6 Merge pull request #1267 from horoc/go-server-rocketmq-message-convert
56ba18cc is described below
commit 56ba18cc56793b66cb979b5aacfa9086a8fd38b5
Author: horoc <ho...@gmail.com>
AuthorDate: Wed Sep 14 17:11:37 2022 +0800
add rocketmq message convert
---
.../rocketmq/constants/message_constants.go | 25 +++++
.../rocketmq/convert/rocketmq_message_reader.go | 110 +++++++++++++++++++++
.../convert/rocketmq_message_reader_test.go | 47 +++++++++
.../rocketmq/convert/rocketmq_message_writer.go | 86 ++++++++++++++++
.../convert/rocketmq_message_writer_test.go | 44 +++++++++
5 files changed, 312 insertions(+)
diff --git a/eventmesh-server-go/plugin/connector/rocketmq/constants/message_constants.go b/eventmesh-server-go/plugin/connector/rocketmq/constants/message_constants.go
new file mode 100644
index 00000000..371cbec1
--- /dev/null
+++ b/eventmesh-server-go/plugin/connector/rocketmq/constants/message_constants.go
@@ -0,0 +1,25 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package constants
+
+const (
+ CloudEventMessageType = "eventmeshmessage"
+ RocketMQMessageContentTypeProperties = "datacontenttype"
+ MessagePropertySeparator = "99"
+ ConsumerGroupBroadcastPrefix = "broadcast"
+ PropertyMessageBornTimestamp = "borntimestamp"
+ PropertyMessageStoreTimestamp = "storetimestamp"
+)
diff --git a/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_reader.go b/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_reader.go
new file mode 100644
index 00000000..8d22baaa
--- /dev/null
+++ b/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_reader.go
@@ -0,0 +1,110 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package convert
+
+import (
+ "bytes"
+ "context"
+ "github.com/apache/incubator-eventmesh/eventmesh-server-go/plugin/connector/rocketmq/constants"
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+ ce "github.com/cloudevents/sdk-go/v2"
+ "github.com/cloudevents/sdk-go/v2/binding"
+ "github.com/cloudevents/sdk-go/v2/binding/spec"
+ "github.com/cloudevents/sdk-go/v2/event"
+)
+
+// RocketMQMessageReader cloud event message reader
+type RocketMQMessageReader struct {
+ topic string
+ properties map[string]string
+ body []byte
+ version spec.Version
+}
+
+// NewRocketMQMessageReader get RocketMQ message reader
+func NewRocketMQMessageReader(message *primitive.Message) *RocketMQMessageReader {
+ return &RocketMQMessageReader{
+ topic: message.Topic,
+ properties: message.GetProperties(),
+ body: message.Body,
+ version: spec.VS.Version(event.CloudEventsVersionV1),
+ }
+}
+
+// ReadEncoding the RocketMQMessageReader only supports binary encoding
+func (r *RocketMQMessageReader) ReadEncoding() binding.Encoding {
+ return binding.EncodingBinary
+}
+
+// ReadBinary read message and write through encoder
+func (r *RocketMQMessageReader) ReadBinary(ctx context.Context, encoder binding.BinaryWriter) (err error) {
+ subject := r.version.Attribute("subject")
+ encoder.SetAttribute(subject, r.topic)
+
+ msgType := r.version.Attribute("type")
+ encoder.SetAttribute(msgType, constants.CloudEventMessageType)
+
+ contentType := r.version.Attribute("datacontenttype")
+ encoder.SetAttribute(contentType, r.properties[constants.RocketMQMessageContentTypeProperties])
+
+ for k, v := range r.properties {
+ attr := r.version.Attribute(k)
+ if attr != nil {
+ err = encoder.SetAttribute(attr, v)
+ } else {
+ err = encoder.SetExtension(k, v)
+ }
+ if err != nil {
+ return err
+ }
+ }
+
+ if r.body != nil {
+ err = encoder.SetData(bytes.NewReader(r.body))
+ if err != nil {
+ return err
+ }
+ }
+ return
+}
+
+// GetAttribute get attribute
+func (r *RocketMQMessageReader) GetAttribute(k spec.Kind) (spec.Attribute, interface{}) {
+ attr := r.version.Attribute(k.String())
+ if attr != nil {
+ return attr, r.properties[k.String()]
+ }
+ return attr, nil
+}
+
+// GetExtension get extension
+func (r *RocketMQMessageReader) GetExtension(s string) interface{} {
+ return r.properties[s]
+}
+
+// ReadStructured the RocketMQMessageReader only supports binary encoding
+func (r *RocketMQMessageReader) ReadStructured(context.Context, binding.StructuredWriter) error {
+ return binding.ErrNotStructured
+}
+
+// ToCloudEvent convert RocketMQ message to cloud event
+func (r *RocketMQMessageReader) ToCloudEvent(ctx context.Context) (*ce.Event, error) {
+ event, err := binding.ToEvent(ctx, r)
+ if err != nil {
+ return nil, err
+ }
+ return event, nil
+}
diff --git a/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_reader_test.go b/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_reader_test.go
new file mode 100644
index 00000000..3ac39465
--- /dev/null
+++ b/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_reader_test.go
@@ -0,0 +1,47 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package convert
+
+import (
+ "context"
+ "github.com/apache/incubator-eventmesh/eventmesh-server-go/plugin/connector/rocketmq/constants"
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+ "github.com/cloudevents/sdk-go/v2/event"
+ "github.com/cloudevents/sdk-go/v2/event/datacodec"
+ "github.com/stretchr/testify/require"
+ "testing"
+)
+
+func TestRocketMQMessageReader_ToCloudEvent(t *testing.T) {
+ e, err := NewRocketMQMessageReader(getTestMessage()).ToCloudEvent(context.Background())
+ require.True(t, err == nil)
+ require.True(t, e.ID() == "MockEventId")
+ require.True(t, e.Extensions()["protocoltype"] == "http")
+ data, _ := datacodec.Encode(context.Background(), event.ApplicationJSON, "{\"data\":\"foo\"}")
+ require.True(t, string(e.Data()) == string(data))
+}
+
+func getTestMessage() *primitive.Message {
+ data, _ := datacodec.Encode(context.Background(), event.ApplicationJSON, "{\"data\":\"foo\"}")
+ message := &primitive.Message{
+ Topic: "TopicTest",
+ Body: data,
+ }
+ message.WithProperty("id", "MockEventId")
+ message.WithProperty(constants.RocketMQMessageContentTypeProperties, event.ApplicationJSON)
+ message.WithProperty("protocoltype", "http")
+ return message
+}
diff --git a/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_writer.go b/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_writer.go
new file mode 100644
index 00000000..d6185ff1
--- /dev/null
+++ b/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_writer.go
@@ -0,0 +1,86 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package convert
+
+import (
+ "context"
+ "github.com/apache/rocketmq-client-go/v2/primitive"
+ ce "github.com/cloudevents/sdk-go/v2"
+ "github.com/cloudevents/sdk-go/v2/binding"
+ "github.com/cloudevents/sdk-go/v2/binding/spec"
+ "github.com/cloudevents/sdk-go/v2/types"
+ "io"
+ "io/ioutil"
+)
+
+// RocketMQMessageWriter cloud event message writer
+type RocketMQMessageWriter struct {
+ message *primitive.Message
+}
+
+// NewRocketMQMessageWriter get RocketMQ message writer
+func NewRocketMQMessageWriter(topic string) *RocketMQMessageWriter {
+ message := &primitive.Message{}
+ message.Topic = topic
+ return &RocketMQMessageWriter{message: message}
+}
+
+// SetAttribute set attribute
+func (r *RocketMQMessageWriter) SetAttribute(attribute spec.Attribute, value interface{}) error {
+ val, err := types.Format(value)
+ if err != nil {
+ return err
+ }
+ r.message.WithProperty(attribute.Name(), val)
+ return nil
+}
+
+// SetExtension set extension
+func (r *RocketMQMessageWriter) SetExtension(name string, value interface{}) error {
+ r.message.WithProperty(name, value.(string))
+ return nil
+}
+
+// Start to start writing, do nothing
+func (r *RocketMQMessageWriter) Start(ctx context.Context) error {
+ // No-Op
+ return nil
+}
+
+// SetData set data from reader
+func (r *RocketMQMessageWriter) SetData(data io.Reader) error {
+ b, err := ioutil.ReadAll(data)
+ if err != nil {
+ return nil
+ }
+ r.message.Body = b
+ return nil
+}
+
+// End the end of writing, do nothing
+func (r *RocketMQMessageWriter) End(ctx context.Context) error {
+ // No-Op
+ return nil
+}
+
+// ToMessage convert cloud event to RocketMQ message
+func (r *RocketMQMessageWriter) ToMessage(ctx context.Context, cloudEvent *ce.Event) (*primitive.Message, error) {
+ _, err := binding.Write(binding.WithForceBinary(ctx), (*binding.EventMessage)(cloudEvent), nil, r)
+ if err != nil {
+ return nil, err
+ }
+ return r.message, err
+}
diff --git a/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_writer_test.go b/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_writer_test.go
new file mode 100644
index 00000000..6835b262
--- /dev/null
+++ b/eventmesh-server-go/plugin/connector/rocketmq/convert/rocketmq_message_writer_test.go
@@ -0,0 +1,44 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package convert
+
+import (
+ "context"
+ ce "github.com/cloudevents/sdk-go/v2"
+ "github.com/cloudevents/sdk-go/v2/event"
+ "github.com/cloudevents/sdk-go/v2/event/datacodec"
+ "github.com/stretchr/testify/require"
+ "testing"
+)
+
+func TestWriteBinary(t *testing.T) {
+ message, err := NewRocketMQMessageWriter("test-topic").ToMessage(context.Background(), getTestEvent())
+ require.True(t, err == nil)
+ require.True(t, message.GetProperty("id") == "MockEventId")
+ require.True(t, message.GetProperty("protocoltype") == "http")
+ data, _ := datacodec.Encode(context.Background(), event.ApplicationJSON, "{\"data\":\"foo\"}")
+ require.True(t, string(message.Body) == string(data))
+}
+
+func getTestEvent() *ce.Event {
+ event := ce.NewEvent()
+ event.SetSubject("TopicTest")
+ event.SetID("MockEventId")
+ event.SetDataContentType("application/json")
+ event.SetExtension("protocoltype", "http")
+ event.SetData("application/json", "{\"data\":\"foo\"}")
+ return &event
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org