Expose NumPartitions() method on producer interface and GetHashingFunction() method

Signed-off-by: Chen Liu <[email protected]>

Author: hunter2046

pulsar/producer.go

@@ -19,6 +19,7 @@ package pulsar
 
 import (
 	"context"
+	"github.com/apache/pulsar-client-go/pulsar/internal"
 	"time"
 )
 
@@ -182,6 +183,9 @@ type Producer interface {
 	// the eventual error in publishing
 	SendAsync(context.Context, *ProducerMessage, func(MessageID, *ProducerMessage, error))
 
+	// NumPartitions return the number of partitions that the topic has
+	NumPartitions() uint32
+
 	// LastSequenceID get the last sequence id that was published by this producer.
 	// This represent either the automatically assigned or custom sequence id (set on the ProducerMessage) that
 	// was published and acknowledged by the broker.
@@ -199,3 +203,15 @@ type Producer interface {
 	// of errors, pending writes will not be retried.
 	Close()
 }
+
+// GetHashingFunction return the corresponding hashing function for the hashing scheme
+func GetHashingFunction(s HashingScheme) func(string) uint32 {
+	switch s {
+	case JavaStringHash:
+		return internal.JavaStringHash
+	case Murmur3_32Hash:
+		return internal.Murmur3_32Hash
+	default:
+		return internal.JavaStringHash
+	}
+}

pulsar/producer_impl.go

@@ -59,17 +59,6 @@ type producer struct {
 
 var partitionsAutoDiscoveryInterval = 1 * time.Minute
 
-func getHashingFunction(s HashingScheme) func(string) uint32 {
-	switch s {
-	case JavaStringHash:
-		return internal.JavaStringHash
-	case Murmur3_32Hash:
-		return internal.Murmur3_32Hash
-	default:
-		return internal.JavaStringHash
-	}
-}
-
 func newProducer(client *client, options *ProducerOptions) (*producer, error) {
 	if options.Topic == "" {
 		return nil, newError(InvalidTopicName, "Topic name is required for producer")
@@ -102,7 +91,7 @@ func newProducer(client *client, options *ProducerOptions) (*producer, error) {
 
 	if options.MessageRouter == nil {
 		internalRouter := NewDefaultRouter(
-			getHashingFunction(options.HashingScheme),
+			GetHashingFunction(options.HashingScheme),
 			options.BatchingMaxMessages,
 			options.BatchingMaxSize,
 			options.BatchingMaxPublishDelay,

pulsar/producer_partition.go

@@ -324,6 +324,10 @@ func (p *partitionProducer) Name() string {
 	return p.producerName
 }
 
+func (p *partitionProducer) NumPartitions() uint32 {
+	return 1
+}
+
 func (p *partitionProducer) internalSend(request *sendRequest) {
 	p.log.Debug("Received send request: ", *request)
 

pulsar/producer_test.go

@@ -21,6 +21,8 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"reflect"
+	"runtime"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -41,6 +43,12 @@ func TestInvalidURL(t *testing.T) {
 	}
 }
 
+func TestGetHashingFunction(t *testing.T) {
+	assertHashingFunctionEqual(t, internal.JavaStringHash, GetHashingFunction(-1))
+	assertHashingFunctionEqual(t, internal.JavaStringHash, GetHashingFunction(JavaStringHash))
+	assertHashingFunctionEqual(t, internal.Murmur3_32Hash, GetHashingFunction(Murmur3_32Hash))
+}
+
 func TestProducerConnectError(t *testing.T) {
 	client, err := NewClient(ClientOptions{
 		URL: "pulsar://invalid-hostname:6650",
@@ -94,6 +102,7 @@ func TestSimpleProducer(t *testing.T) {
 
 	assert.NoError(t, err)
 	assert.NotNil(t, producer)
+	assert.Equal(t, uint32(1), producer.NumPartitions())
 	defer producer.Close()
 
 	for i := 0; i < 10; i++ {
@@ -398,6 +407,7 @@ func TestFlushInPartitionedProducer(t *testing.T) {
 		BatchingMaxPublishDelay: time.Second * 10,
 	})
 	assert.Nil(t, err)
+	assert.Equal(t, uint32(numberOfPartitions), producer.NumPartitions())
 	defer producer.Close()
 
 	// send 5 messages
@@ -473,6 +483,7 @@ func TestRoundRobinRouterPartitionedProducer(t *testing.T) {
 		DisableBatching: true,
 	})
 	assert.Nil(t, err)
+	assert.Equal(t, uint32(numberOfPartitions), producer.NumPartitions())
 	defer producer.Close()
 
 	// send 5 messages
@@ -1031,3 +1042,9 @@ func TestProducerWithInterceptors(t *testing.T) {
 	assert.Equal(t, 10, metric.sendn)
 	assert.Equal(t, 10, metric.ackn)
 }
+
+func assertHashingFunctionEqual(t *testing.T, func1, func2 func(string) uint32) {
+	funcName1 := runtime.FuncForPC(reflect.ValueOf(func1).Pointer()).Name()
+	funcName2 := runtime.FuncForPC(reflect.ValueOf(func2).Pointer()).Name()
+	assert.Equal(t, funcName1, funcName2)
+}