Add concurrency parameter to limit parallel callback functions

buraksezer · Aug 1, 2020 · 4fdd7c3 · 4fdd7c3
1 parent e4fd637
commit 4fdd7c3
Show file tree

Hide file tree

Showing 2 changed files with 19 additions and 16 deletions.
diff --git a/client/dtopic.go b/client/dtopic.go
@@ -28,26 +28,28 @@ import (
 // DTopic denotes a distributed topic instance in the cluster.
 type DTopic struct {
 	*Client
-	name string
-	flag int16
+	name        string
+	flag        int16
+	concurrency int
 
 	mu        sync.Mutex
 	listeners map[uint64]struct{}
 }
 
 // NewDTopic creates and returns a new DTopic instance. It doesn't create any underlying TCP socket and goroutine.
-func (c *Client) NewDTopic(name string, flag int16) (*DTopic, error) {
+func (c *Client) NewDTopic(name string, concurrency int, flag int16) (*DTopic, error) {
 	if flag&olric.UnorderedDelivery == 0 && flag&olric.OrderedDelivery == 0 {
 		return nil, fmt.Errorf("invalid delivery mode: %w", olric.ErrInvalidArgument)
 	}
 	if flag&olric.OrderedDelivery != 0 {
 		return nil, olric.ErrNotImplemented
 	}
 	return &DTopic{
-		Client:    c,
-		name:      name,
-		flag:      flag,
-		listeners: make(map[uint64]struct{}),
+		Client:      c,
+		name:        name,
+		flag:        flag,
+		concurrency: concurrency,
+		listeners:   make(map[uint64]struct{}),
 	}, nil
 }
 
@@ -71,9 +73,10 @@ func (dt *DTopic) listen(l *listener, f func(olric.DTopicMessage)) {
 	defer dt.wg.Done()
 
 	// Limit concurrency to prevent CPU or network I/O starvation.
-	// TODO: Configure this
-	num := int64(runtime.NumCPU())
-	sem := semaphore.NewWeighted(num)
+	if dt.concurrency == 0 {
+		dt.concurrency = runtime.NumCPU()
+	}
+	sem := semaphore.NewWeighted(int64(dt.concurrency))
 	for {
 		select {
 		case <-l.ctx.Done():

diff --git a/client/dtopic_test.go b/client/dtopic_test.go
@@ -41,7 +41,7 @@ func TestClient_DTopicPublish(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
-	dt, err := c.NewDTopic("my-dtopic", olric.UnorderedDelivery)
+	dt, err := c.NewDTopic("my-dtopic", 0, olric.UnorderedDelivery)
 	if err != nil {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
@@ -68,7 +68,7 @@ func TestClient_DTopicPublishMessages(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
-	dt, err := c.NewDTopic("my-dtopic", olric.UnorderedDelivery)
+	dt, err := c.NewDTopic("my-dtopic", 0, olric.UnorderedDelivery)
 	if err != nil {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
@@ -123,7 +123,7 @@ func TestClient_DTopicAddListener(t *testing.T) {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
 	onMessage := func(message olric.DTopicMessage) {}
-	dt, err := c.NewDTopic("my-dtopic", olric.UnorderedDelivery)
+	dt, err := c.NewDTopic("my-dtopic", 0, olric.UnorderedDelivery)
 	if err != nil {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
@@ -161,7 +161,7 @@ func TestClient_DTopicOnMessage(t *testing.T) {
 		}
 	}
 
-	dt, err := c.NewDTopic("my-dtopic", olric.UnorderedDelivery)
+	dt, err := c.NewDTopic("my-dtopic", 0, olric.UnorderedDelivery)
 	if err != nil {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
@@ -200,7 +200,7 @@ func TestClient_DTopicRemoveListener(t *testing.T) {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
 	onMessage := func(message olric.DTopicMessage) {}
-	dt, err := c.NewDTopic("my-dtopic", olric.UnorderedDelivery)
+	dt, err := c.NewDTopic("my-dtopic", 0, olric.UnorderedDelivery)
 	if err != nil {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
@@ -242,7 +242,7 @@ func TestClient_DTopicDestroy(t *testing.T) {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}
 	onMessage := func(message olric.DTopicMessage) {}
-	dt, err := c.NewDTopic("my-dtopic", olric.UnorderedDelivery)
+	dt, err := c.NewDTopic("my-dtopic", 0, olric.UnorderedDelivery)
 	if err != nil {
 		t.Fatalf("Expected nil. Got: %v", err)
 	}