Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support listener name for go client #502

Merged
merged 3 commits into from
Apr 6, 2021
Merged

Support listener name for go client #502

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
87101db
Support listener name for go client
wolfstudy Apr 6, 2021
12eb924
fix test error
wolfstudy Apr 6, 2021
c7f089a
fix test error
wolfstudy Apr 6, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions pulsar/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,9 @@ type ClientOptions struct {
// Configure whether the Pulsar client verify the validity of the host name from broker (default: false)
TLSValidateHostname bool

// Configure the net model for vpc user to connect the pulsar broker
ListenerName string

// Max number of connections to a single broker that will kept in the pool. (Default: 1 connection)
MaxConnectionsPerBroker int

Expand Down
3 changes: 2 additions & 1 deletion pulsar/client_impl.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,8 @@ func newClient(options ClientOptions) (Client, error) {
serviceNameResolver := internal.NewPulsarServiceNameResolver(url)

c.rpcClient = internal.NewRPCClient(url, serviceNameResolver, c.cnxPool, operationTimeout, logger, metrics)
c.lookupService = internal.NewLookupService(c.rpcClient, url, serviceNameResolver, tlsConfig != nil, logger, metrics)
c.lookupService = internal.NewLookupService(c.rpcClient, url, serviceNameResolver,
tlsConfig != nil, options.ListenerName, logger, metrics)
c.handlers = internal.NewClientHandlers()

return c, nil
Expand Down
24 changes: 15 additions & 9 deletions pulsar/internal/lookup_service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,21 +46,25 @@ type LookupService interface {

type lookupService struct {
rpcClient RPCClient
serviceNameResolver ServiceNameResolver
serviceURL *url.URL
wolfstudy marked this conversation as resolved.
Show resolved Hide resolved
tlsEnabled bool
listenerName string
log log.Logger
metrics *Metrics
serviceNameResolver ServiceNameResolver
}

// NewLookupService init a lookup service struct and return an object of LookupService.
func NewLookupService(rpcClient RPCClient, serviceURL *url.URL, serviceNameResolver ServiceNameResolver,
tlsEnabled bool, logger log.Logger, metrics *Metrics) LookupService {
tlsEnabled bool, listenerName string, logger log.Logger, metrics *Metrics) LookupService {
return &lookupService{
rpcClient: rpcClient,
serviceNameResolver: serviceNameResolver,
serviceURL: serviceURL,
tlsEnabled: tlsEnabled,
listenerName: listenerName,
log: logger.SubLogger(log.Fields{"serviceURL": serviceURL}),
metrics: metrics,
serviceNameResolver: serviceNameResolver,
}
}

Expand Down Expand Up @@ -96,9 +100,10 @@ func (ls *lookupService) Lookup(topic string) (*LookupResult, error) {
ls.metrics.LookupRequestsCount.Inc()
id := ls.rpcClient.NewRequestID()
res, err := ls.rpcClient.RequestToAnyBroker(id, pb.BaseCommand_LOOKUP, &pb.CommandLookupTopic{
RequestId: &id,
Topic: &topic,
Authoritative: proto.Bool(false),
RequestId: &id,
Topic: &topic,
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(ls.listenerName),
})
if err != nil {
return nil, err
Expand All @@ -120,9 +125,10 @@ func (ls *lookupService) Lookup(topic string) (*LookupResult, error) {

id := ls.rpcClient.NewRequestID()
res, err = ls.rpcClient.Request(logicalAddress, physicalAddr, id, pb.BaseCommand_LOOKUP, &pb.CommandLookupTopic{
RequestId: &id,
Topic: &topic,
Authoritative: lr.Authoritative,
RequestId: &id,
Topic: &topic,
Authoritative: lr.Authoritative,
AdvertisedListenerName: proto.String(ls.listenerName),
})
if err != nil {
return nil, err
Expand Down
90 changes: 50 additions & 40 deletions pulsar/internal/lookup_service_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,9 +119,10 @@ func TestLookupSuccess(t *testing.T) {

expectedRequests: []pb.CommandLookupTopic{
{
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(""),
},
},
mockedResponses: []pb.CommandLookupTopicResponse{
Expand All @@ -132,7 +133,7 @@ func TestLookupSuccess(t *testing.T) {
BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
},
},
}, url, serviceNameResolver, false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.NoError(t, err)
Expand All @@ -152,9 +153,10 @@ func TestTlsLookupSuccess(t *testing.T) {

expectedRequests: []pb.CommandLookupTopic{
{
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(""),
},
},
mockedResponses: []pb.CommandLookupTopicResponse{
Expand All @@ -165,7 +167,7 @@ func TestTlsLookupSuccess(t *testing.T) {
BrokerServiceUrlTls: proto.String("pulsar+ssl://broker-1:6651"),
},
},
}, url, serviceNameResolver, true, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, serviceNameResolver, true, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.NoError(t, err)
Expand All @@ -185,9 +187,10 @@ func TestLookupWithProxy(t *testing.T) {

expectedRequests: []pb.CommandLookupTopic{
{
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(""),
},
},
mockedResponses: []pb.CommandLookupTopicResponse{
Expand All @@ -199,7 +202,7 @@ func TestLookupWithProxy(t *testing.T) {
ProxyThroughServiceUrl: proto.Bool(true),
},
},
}, url, serviceNameResolver, false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.NoError(t, err)
Expand All @@ -218,9 +221,10 @@ func TestTlsLookupWithProxy(t *testing.T) {

expectedRequests: []pb.CommandLookupTopic{
{
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(""),
},
},
mockedResponses: []pb.CommandLookupTopicResponse{
Expand All @@ -232,7 +236,7 @@ func TestTlsLookupWithProxy(t *testing.T) {
ProxyThroughServiceUrl: proto.Bool(true),
},
},
}, url, NewPulsarServiceNameResolver(url), true, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, NewPulsarServiceNameResolver(url), true, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.NoError(t, err)
Expand All @@ -252,14 +256,16 @@ func TestLookupWithRedirect(t *testing.T) {

expectedRequests: []pb.CommandLookupTopic{
{
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(""),
},
{
RequestId: proto.Uint64(2),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(true),
RequestId: proto.Uint64(2),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(true),
AdvertisedListenerName: proto.String(""),
},
},
mockedResponses: []pb.CommandLookupTopicResponse{
Expand All @@ -276,7 +282,7 @@ func TestLookupWithRedirect(t *testing.T) {
BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
},
},
}, url, NewPulsarServiceNameResolver(url), false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.NoError(t, err)
Expand All @@ -296,14 +302,16 @@ func TestTlsLookupWithRedirect(t *testing.T) {

expectedRequests: []pb.CommandLookupTopic{
{
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(""),
},
{
RequestId: proto.Uint64(2),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(true),
RequestId: proto.Uint64(2),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(true),
AdvertisedListenerName: proto.String(""),
},
},
mockedResponses: []pb.CommandLookupTopicResponse{
Expand All @@ -320,7 +328,7 @@ func TestTlsLookupWithRedirect(t *testing.T) {
BrokerServiceUrlTls: proto.String("pulsar+ssl://broker-1:6651"),
},
},
}, url, NewPulsarServiceNameResolver(url), true, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, NewPulsarServiceNameResolver(url), true, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.NoError(t, err)
Expand All @@ -339,9 +347,10 @@ func TestLookupWithInvalidUrlResponse(t *testing.T) {

expectedRequests: []pb.CommandLookupTopic{
{
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(""),
},
},
mockedResponses: []pb.CommandLookupTopicResponse{
Expand All @@ -353,7 +362,7 @@ func TestLookupWithInvalidUrlResponse(t *testing.T) {
ProxyThroughServiceUrl: proto.Bool(false),
},
},
}, url, NewPulsarServiceNameResolver(url), false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.Error(t, err)
Expand All @@ -369,9 +378,10 @@ func TestLookupWithLookupFailure(t *testing.T) {

expectedRequests: []pb.CommandLookupTopic{
{
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
RequestId: proto.Uint64(1),
Topic: proto.String("my-topic"),
Authoritative: proto.Bool(false),
AdvertisedListenerName: proto.String(""),
},
},
mockedResponses: []pb.CommandLookupTopicResponse{
Expand All @@ -381,7 +391,7 @@ func TestLookupWithLookupFailure(t *testing.T) {
Authoritative: proto.Bool(true),
},
},
}, url, NewPulsarServiceNameResolver(url), false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.Error(t, err)
Expand Down Expand Up @@ -468,7 +478,7 @@ func TestGetPartitionedTopicMetadataSuccess(t *testing.T) {
Response: pb.CommandPartitionedTopicMetadataResponse_Success.Enum(),
},
},
}, url, serviceNameResolver, false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

metadata, err := ls.GetPartitionedTopicMetadata("my-topic")
assert.NoError(t, err)
Expand Down Expand Up @@ -499,7 +509,7 @@ func TestLookupSuccessWithMultipleHosts(t *testing.T) {
BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
},
},
}, url, serviceNameResolver, false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))

lr, err := ls.Lookup("my-topic")
assert.NoError(t, err)
Expand Down
Loading