feat(event client): add WithEventConsumerTimeout option (#186)

Signed-off-by: Alex Klakovsky <klakovskiy@gmail.com>

Author: klakovsky

pkg/client/event/event.go

@@ -15,21 +15,25 @@ SPDX-License-Identifier: Apache-2.0
 package event
 
 import (
+	"time"
+
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/options"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/client"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient/seek"
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/service/dispatcher"
 	"github.com/pkg/errors"
 )
 
 // Client enables access to a channel events on a Fabric network.
 type Client struct {
-	eventService      fab.EventService
-	permitBlockEvents bool
-	fromBlock         uint64
-	seekType          seek.Type
+	eventService         fab.EventService
+	permitBlockEvents    bool
+	fromBlock            uint64
+	seekType             seek.Type
+	eventConsumerTimeout *time.Duration
 }
 
 // New returns a Client instance. Client receives events such as block, filtered block,
@@ -64,6 +68,9 @@ func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client
 				opts = append(opts, deliverclient.WithBlockNum(eventClient.fromBlock))
 			}
 		}
+		if eventClient.eventConsumerTimeout != nil {
+			opts = append(opts, dispatcher.WithEventConsumerTimeout(*eventClient.eventConsumerTimeout))
+		}
 		es, err = channelContext.ChannelService().EventService(opts...)
 	} else {
 		es, err = channelContext.ChannelService().EventService()

pkg/client/event/event_test.go

@@ -43,7 +43,7 @@ func TestNewEventClient(t *testing.T) {
 		t.Fatalf("Failed to create new event client: %s", err)
 	}
 
-	_, err = New(ctx, WithBlockEvents(), WithSeekType(seek.Newest), WithBlockNum(math.MaxUint64))
+	_, err = New(ctx, WithBlockEvents(), WithSeekType(seek.Newest), WithBlockNum(math.MaxUint64), WithEventConsumerTimeout(500 * time.Millisecond))
 	if err != nil {
 		t.Fatalf("Failed to create new event client: %s", err)
 	}

pkg/client/event/opts.go

@@ -6,7 +6,11 @@ SPDX-License-Identifier: Apache-2.0
 
 package event
 
-import "github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient/seek"
+import (
+	"time"
+
+	"github.com/hyperledger/fabric-sdk-go/pkg/fab/events/deliverclient/seek"
+)
 
 // ClientOption describes a functional parameter for the New constructor
 type ClientOption func(*Client) error
@@ -37,3 +41,14 @@ func WithSeekType(seek seek.Type) ClientOption {
 		return nil
 	}
 }
+
+// WithEventConsumerTimeout is the timeout when sending events to a registered consumer.
+// If < 0, if buffer full, unblocks immediately and does not send.
+// If 0, if buffer full, will block and guarantee the event will be sent out.
+// If > 0, if buffer full, blocks util timeout.
+func WithEventConsumerTimeout(value time.Duration) ClientOption {
+	return func(c *Client) error {
+		c.eventConsumerTimeout = &value
+		return nil
+	}
+}