This is an automated email from the ASF dual-hosted git repository.
jonyang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 5354032c0 fix issue2885 (#2887)
5354032c0 is described below
commit 5354032c038c76a9eee17df6846c61b045088184
Author: jonyangx <[email protected]>
AuthorDate: Tue Jan 10 23:45:01 2023 +0800
fix issue2885 (#2887)
Co-authored-by: jonyangx <[email protected]>
---
.../http/consumer/consumer_suite_test.go | 62 ++++++++++
.../http/consumer/eventmesh_http_consumer.go | 12 ++
.../http/consumer/eventmesh_http_consumer_test.go | 136 +++++++++------------
3 files changed, 133 insertions(+), 77 deletions(-)
diff --git a/eventmesh-sdk-go/http/consumer/consumer_suite_test.go
b/eventmesh-sdk-go/http/consumer/consumer_suite_test.go
new file mode 100644
index 000000000..1d2b8f786
--- /dev/null
+++ b/eventmesh-sdk-go/http/consumer/consumer_suite_test.go
@@ -0,0 +1,62 @@
+/**
+ * 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 consumer
+
+import (
+ "fmt"
+ "github.com/apache/incubator-eventmesh/eventmesh-sdk-go/http/conf"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
+)
+
+func TestConsumerAPIs(t *testing.T) {
+ RegisterFailHandler(Fail)
+ RunSpecs(t, "consumer module Tests")
+}
+
+var server *httptest.Server
+var eventMeshHttpConsumer *EventMeshHttpConsumer
+
+var _ = BeforeSuite(func() {
+ f := func(w http.ResponseWriter, r *http.Request) {
+ w.WriteHeader(http.StatusOK)
+ w.Write([]byte(`{"retCode":0}`))
+ }
+ server = httptest.NewServer(http.HandlerFunc(f))
+
+ eventMeshClientConfig := conf.DefaultEventMeshHttpClientConfig
+ sp := strings.Split(server.URL, ":")
+ eventMeshClientConfig.SetLiteEventMeshAddr(fmt.Sprintf("127.0.0.1:%s",
sp[len(sp)-1]))
+ eventMeshHttpConsumer = NewEventMeshHttpConsumer(eventMeshClientConfig)
+
+})
+
+var _ = AfterSuite(func() {
+ if server != nil {
+ server.Close()
+ }
+
+ if eventMeshHttpConsumer != nil {
+ eventMeshHttpConsumer.Close()
+ }
+})
diff --git a/eventmesh-sdk-go/http/consumer/eventmesh_http_consumer.go
b/eventmesh-sdk-go/http/consumer/eventmesh_http_consumer.go
index 21a37df11..678b37b69 100644
--- a/eventmesh-sdk-go/http/consumer/eventmesh_http_consumer.go
+++ b/eventmesh-sdk-go/http/consumer/eventmesh_http_consumer.go
@@ -95,6 +95,18 @@ func (e *EventMeshHttpConsumer) Subscribe(topicList
[]protocol.SubscriptionItem,
e.mutex.Lock()
defer e.mutex.Unlock()
e.subscriptions = append(e.subscriptions, topicList...)
+
+ result := make([]protocol.SubscriptionItem, 0, len(e.subscriptions))
+ tmpMap := make(map[string]struct{})
+ for _, item := range e.subscriptions {
+ if _, ok := tmpMap[item.Topic]; !ok {
+ tmpMap[item.Topic] = struct{}{}
+ result = append(result, item)
+ }
+ }
+
+ e.subscriptions = result
+
}
func (e *EventMeshHttpConsumer) Unsubscribe(topicList []string, subscribeUrl
string) {
diff --git a/eventmesh-sdk-go/http/consumer/eventmesh_http_consumer_test.go
b/eventmesh-sdk-go/http/consumer/eventmesh_http_consumer_test.go
index 7d427b8df..3acdd975c 100644
--- a/eventmesh-sdk-go/http/consumer/eventmesh_http_consumer_test.go
+++ b/eventmesh-sdk-go/http/consumer/eventmesh_http_consumer_test.go
@@ -1,91 +1,73 @@
-// 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.
+/**
+ * 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 consumer
import (
- "fmt"
"github.com/apache/incubator-eventmesh/eventmesh-sdk-go/common/protocol"
- "github.com/apache/incubator-eventmesh/eventmesh-sdk-go/http/conf"
- "github.com/stretchr/testify/assert"
- "net/http"
- "net/http/httptest"
- "strings"
- "testing"
+ . "github.com/onsi/ginkgo"
+ . "github.com/onsi/gomega"
)
-func TestEventMeshHttpConsumer_Subscribe(t *testing.T) {
- f := func(w http.ResponseWriter, r *http.Request) {
- w.WriteHeader(http.StatusOK)
- w.Write([]byte(`{"retCode":0}`))
- }
- server := httptest.NewServer(http.HandlerFunc(f))
- defer server.Close()
+var _ = Describe("EventMeshHttpConsumer test", func() {
- eventMeshClientConfig := conf.DefaultEventMeshHttpClientConfig
- sp := strings.Split(server.URL, ":")
- eventMeshClientConfig.SetLiteEventMeshAddr(fmt.Sprintf("127.0.0.1:%s",
sp[len(sp)-1]))
- eventMeshHttpConsumer := NewEventMeshHttpConsumer(eventMeshClientConfig)
+ Context("Subscribe test ", func() {
+ It("should success", func() {
- subscribeUrl := "http://mock-url";
- topicList := []protocol.SubscriptionItem{
- {
- Topic: "mock-topic",
- Mode: protocol.DefaultSubscriptionMode.CLUSTERING,
- Type: protocol.DefaultSubscriptionType.ASYNC,
- },
- }
- eventMeshHttpConsumer.Subscribe(topicList, subscribeUrl)
- assert.Equal(t, 1, len(eventMeshHttpConsumer.subscriptions))
+ subscribeUrl := "http://mock-url";
+ topicList := []protocol.SubscriptionItem{
+ {
+ Topic: "mock-topic",
+ Mode:
protocol.DefaultSubscriptionMode.CLUSTERING,
+ Type:
protocol.DefaultSubscriptionType.ASYNC,
+ },
+ }
+ eventMeshHttpConsumer.Subscribe(topicList, subscribeUrl)
- topicList = []protocol.SubscriptionItem{
- {
- Topic: "mock-topic2",
- Mode: protocol.DefaultSubscriptionMode.CLUSTERING,
- Type: protocol.DefaultSubscriptionType.ASYNC,
- },
- }
- eventMeshHttpConsumer.Subscribe(topicList, subscribeUrl)
- assert.Equal(t, 2, len(eventMeshHttpConsumer.subscriptions))
-}
+ Ω(1).To(Equal(len(eventMeshHttpConsumer.subscriptions)))
-func TestEventMeshHttpConsumer_Unsubscribe(t *testing.T) {
- f := func(w http.ResponseWriter, r *http.Request) {
- w.WriteHeader(http.StatusOK)
- w.Write([]byte(`{"retCode":0}`))
- }
- server := httptest.NewServer(http.HandlerFunc(f))
- defer server.Close()
+ topicList = []protocol.SubscriptionItem{
+ {
+ Topic: "mock-topic2",
+ Mode:
protocol.DefaultSubscriptionMode.CLUSTERING,
+ Type:
protocol.DefaultSubscriptionType.ASYNC,
+ },
+ }
+ eventMeshHttpConsumer.Subscribe(topicList, subscribeUrl)
+ Ω(2).To(Equal(len(eventMeshHttpConsumer.subscriptions)))
+ eventMeshHttpConsumer.Subscribe(topicList, subscribeUrl)
+ Ω(2).To(Equal(len(eventMeshHttpConsumer.subscriptions)))
+ })
- eventMeshClientConfig := conf.DefaultEventMeshHttpClientConfig
- sp := strings.Split(server.URL, ":")
- eventMeshClientConfig.SetLiteEventMeshAddr(fmt.Sprintf("127.0.0.1:%s",
sp[len(sp)-1]))
- eventMeshHttpConsumer := NewEventMeshHttpConsumer(eventMeshClientConfig)
+ })
- subscribeUrl := "http://mock-url";
- subscribeList := []protocol.SubscriptionItem{
- {
- Topic: "mock-topic",
- Mode: protocol.DefaultSubscriptionMode.CLUSTERING,
- Type: protocol.DefaultSubscriptionType.ASYNC,
- },
- }
- eventMeshHttpConsumer.Subscribe(subscribeList, subscribeUrl)
- assert.Equal(t, 1, len(eventMeshHttpConsumer.subscriptions))
+ Context("Unsubscribe test ", func() {
+ It("should success", func() {
+ subscribeUrl := "http://mock-url";
+ var topicList = []string{"mock-topic"}
+ eventMeshHttpConsumer.Unsubscribe(topicList,
subscribeUrl)
+ Ω(1).To(Equal(len(eventMeshHttpConsumer.subscriptions)))
+ topicList = []string{"mock-topic2"}
+ eventMeshHttpConsumer.Unsubscribe(topicList,
subscribeUrl)
+ Ω(0).To(Equal(len(eventMeshHttpConsumer.subscriptions)))
+ topicList = []string{"mock-topic2"}
+ eventMeshHttpConsumer.Unsubscribe(topicList,
subscribeUrl)
+ Ω(0).To(Equal(len(eventMeshHttpConsumer.subscriptions)))
+ })
- var topicList = []string{"mock-topic"}
- eventMeshHttpConsumer.Unsubscribe(topicList, subscribeUrl)
- assert.Equal(t, 0, len(eventMeshHttpConsumer.subscriptions))
-}
+ })
+})
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
