Use Buffered channels to avoid blocking on callback (apache#5336)

ascentstream · Oct 9, 2019 · 5df6488 · 5df6488
1 parent db15d32
commit 5df6488
Show file tree

Hide file tree

Showing 4 changed files with 14 additions and 14 deletions.
diff --git a/pulsar-client-go/pulsar/c_client.go b/pulsar-client-go/pulsar/c_client.go
@@ -194,7 +194,7 @@ func (client *client) CreateProducer(options ProducerOptions) (Producer, error)
 	c := make(chan struct {
 		Producer
 		error
-	})
+	}, 1)
 
 	client.CreateProducerAsync(options, nil, func(producer Producer, err error) {
 		c <- struct {
@@ -214,7 +214,7 @@ func (client *client) CreateProducerWithSchema(options ProducerOptions, schema S
 	c := make(chan struct {
 		Producer
 		error
-	})
+	}, 1)
 
 	client.CreateProducerAsync(options, schema, func(producer Producer, err error) {
 		c <- struct {
@@ -236,7 +236,7 @@ func (client *client) Subscribe(options ConsumerOptions) (Consumer, error) {
 	c := make(chan struct {
 		Consumer
 		error
-	})
+	}, 1)
 
 	client.SubscribeAsync(options, nil, func(consumer Consumer, err error) {
 		c <- struct {
@@ -254,7 +254,7 @@ func (client *client) SubscribeWithSchema(options ConsumerOptions, schema Schema
 	c := make(chan struct {
 		Consumer
 		error
-	})
+	}, 1)
 
 	client.SubscribeAsync(options, schema, func(consumer Consumer, err error) {
 		c <- struct {
@@ -276,7 +276,7 @@ func (client *client) CreateReader(options ReaderOptions) (Reader, error) {
 	c := make(chan struct {
 		Reader
 		error
-	})
+	}, 1)
 
 	client.CreateReaderAsync(options, nil, func(reader Reader, err error) {
 		c <- struct {
@@ -294,7 +294,7 @@ func (client *client) CreateReaderWithSchema(options ReaderOptions, schema Schem
 	c := make(chan struct {
 		Reader
 		error
-	})
+	}, 1)
 
 	client.CreateReaderAsync(options, schema, func(reader Reader, err error) {
 		c <- struct {
@@ -335,7 +335,7 @@ func (client *client) TopicPartitions(topic string) ([]string, error) {
 	c := make(chan struct {
 		partitions []string
 		err        error
-	})
+	}, 1)
 
 	topicPartitionsAsync(client, topic, func(partitions []string, err error) {
 		c <- struct {

diff --git a/pulsar-client-go/pulsar/c_consumer.go b/pulsar-client-go/pulsar/c_consumer.go
@@ -256,7 +256,7 @@ func (c *consumer) Subscription() string {
 }
 
 func (c *consumer) Unsubscribe() error {
-	channel := make(chan error)
+	channel := make(chan error, 1)
 	c.UnsubscribeAsync(func(err error) {
 		channel <- err
 		close(channel)
@@ -320,7 +320,7 @@ func (c *consumer) NackID(msgId MessageID) error {
 }
 
 func (c *consumer) Close() error {
-	channel := make(chan error)
+	channel := make(chan error, 1)
 	c.CloseAsync(func(err error) { channel <- err; close(channel) })
 	return <-channel
 }
@@ -349,7 +349,7 @@ func (c *consumer) RedeliverUnackedMessages() {
 }
 
 func (c *consumer) Seek(msgID MessageID) error {
-	channel := make(chan error)
+	channel := make(chan error, 1)
 	c.SeekAsync(msgID, func(err error) {
 		channel <- err
 		close(channel)

diff --git a/pulsar-client-go/pulsar/c_producer.go b/pulsar-client-go/pulsar/c_producer.go
@@ -230,7 +230,7 @@ func (p *producer) LastSequenceID() int64 {
 }
 
 func (p *producer) Send(ctx context.Context, msg ProducerMessage) error {
-	c := make(chan error)
+	c := make(chan error, 1)
 	p.SendAsync(ctx, msg, func(msg ProducerMessage, err error) { c <- err; close(c) })
 
 	select {
@@ -283,7 +283,7 @@ func (p *producer) SendAsync(ctx context.Context, msg ProducerMessage, callback
 }
 
 func (p *producer) Close() error {
-	c := make(chan error)
+	c := make(chan error, 1)
 	p.CloseAsync(func(err error) { c <- err; close(c) })
 	return <-c
 }
@@ -304,7 +304,7 @@ func pulsarProducerCloseCallbackProxy(res C.pulsar_result, ctx unsafe.Pointer) {
 }
 
 func (p *producer) Flush() error {
-	f := make(chan error)
+	f := make(chan error, 1)
 	p.FlushAsync(func(err error) {
 		f <- err
 		close(f)

diff --git a/pulsar-client-go/pulsar/c_reader.go b/pulsar-client-go/pulsar/c_reader.go
@@ -170,7 +170,7 @@ func (r *reader) HasNext() (bool, error) {
 }
 
 func (r *reader) Close() error {
-	channel := make(chan error)
+	channel := make(chan error, 1)
 	r.CloseAsync(func(err error) { channel <- err; close(channel) })
 	return <-channel
 }