Skip to content

Commit f17deac

Browse files
wolfstudywolfstudy
authored
Support listener name for go client (#502)
* Support listener name for go client Signed-off-by: xiaolongran <xiaolongran@tencent.com> * fix test error Signed-off-by: xiaolongran <xiaolongran@tencent.com> * fix test error Signed-off-by: xiaolongran <xiaolongran@tencent.com> Co-authored-by: xiaolongran <xiaolongran@tencent.com>
1 parent 6ab17dc commit f17deac

6 files changed

Lines changed: 708 additions & 676 deletions

File tree

‎pulsar/client.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -101,6 +101,9 @@ type ClientOptions struct {
101101
// Configure whether the Pulsar client verify the validity of the host name from broker (default: false)
102102
TLSValidateHostname bool
103103

104+
// Configure the net model for vpc user to connect the pulsar broker
105+
ListenerName string
106+
104107
// Max number of connections to a single broker that will kept in the pool. (Default: 1 connection)
105108
MaxConnectionsPerBroker int
106109

‎pulsar/client_impl.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -126,7 +126,8 @@ func newClient(options ClientOptions) (Client, error) {
126126
serviceNameResolver := internal.NewPulsarServiceNameResolver(url)
127127

128128
c.rpcClient = internal.NewRPCClient(url, serviceNameResolver, c.cnxPool, operationTimeout, logger, metrics)
129-
c.lookupService = internal.NewLookupService(c.rpcClient, url, serviceNameResolver, tlsConfig != nil, logger, metrics)
129+
c.lookupService = internal.NewLookupService(c.rpcClient, url, serviceNameResolver,
130+
tlsConfig != nil, options.ListenerName, logger, metrics)
130131
c.handlers = internal.NewClientHandlers()
131132

132133
return c, nil

‎pulsar/internal/lookup_service.go‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -48,19 +48,21 @@ type lookupService struct {
4848
rpcClient RPCClient
4949
serviceNameResolver ServiceNameResolver
5050
tlsEnabled bool
51+
listenerName string
5152
log log.Logger
5253
metrics *Metrics
5354
}
5455

5556
// NewLookupService init a lookup service struct and return an object of LookupService.
5657
func NewLookupService(rpcClient RPCClient, serviceURL *url.URL, serviceNameResolver ServiceNameResolver,
57-
tlsEnabled bool, logger log.Logger, metrics *Metrics) LookupService {
58+
tlsEnabled bool, listenerName string, logger log.Logger, metrics *Metrics) LookupService {
5859
return &lookupService{
5960
rpcClient: rpcClient,
6061
serviceNameResolver: serviceNameResolver,
6162
tlsEnabled: tlsEnabled,
6263
log: logger.SubLogger(log.Fields{"serviceURL": serviceURL}),
6364
metrics: metrics,
65+
listenerName: listenerName,
6466
}
6567
}
6668

@@ -96,9 +98,10 @@ func (ls *lookupService) Lookup(topic string) (*LookupResult, error) {
9698
ls.metrics.LookupRequestsCount.Inc()
9799
id := ls.rpcClient.NewRequestID()
98100
res, err := ls.rpcClient.RequestToAnyBroker(id, pb.BaseCommand_LOOKUP, &pb.CommandLookupTopic{
99-
RequestId: &id,
100-
Topic: &topic,
101-
Authoritative: proto.Bool(false),
101+
RequestId: &id,
102+
Topic: &topic,
103+
Authoritative: proto.Bool(false),
104+
AdvertisedListenerName: proto.String(ls.listenerName),
102105
})
103106
if err != nil {
104107
return nil, err
@@ -120,9 +123,10 @@ func (ls *lookupService) Lookup(topic string) (*LookupResult, error) {
120123

121124
id := ls.rpcClient.NewRequestID()
122125
res, err = ls.rpcClient.Request(logicalAddress, physicalAddr, id, pb.BaseCommand_LOOKUP, &pb.CommandLookupTopic{
123-
RequestId: &id,
124-
Topic: &topic,
125-
Authoritative: lr.Authoritative,
126+
RequestId: &id,
127+
Topic: &topic,
128+
Authoritative: lr.Authoritative,
129+
AdvertisedListenerName: proto.String(ls.listenerName),
126130
})
127131
if err != nil {
128132
return nil, err

‎pulsar/internal/lookup_service_test.go‎

Lines changed: 54 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -119,9 +119,10 @@ func TestLookupSuccess(t *testing.T) {
119119

120120
expectedRequests: []pb.CommandLookupTopic{
121121
{
122-
RequestId: proto.Uint64(1),
123-
Topic: proto.String("my-topic"),
124-
Authoritative: proto.Bool(false),
122+
RequestId: proto.Uint64(1),
123+
Topic: proto.String("my-topic"),
124+
Authoritative: proto.Bool(false),
125+
AdvertisedListenerName: proto.String(""),
125126
},
126127
},
127128
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -132,7 +133,7 @@ func TestLookupSuccess(t *testing.T) {
132133
BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
133134
},
134135
},
135-
}, url, serviceNameResolver, false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
136+
}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
136137

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

153154
expectedRequests: []pb.CommandLookupTopic{
154155
{
155-
RequestId: proto.Uint64(1),
156-
Topic: proto.String("my-topic"),
157-
Authoritative: proto.Bool(false),
156+
RequestId: proto.Uint64(1),
157+
Topic: proto.String("my-topic"),
158+
Authoritative: proto.Bool(false),
159+
AdvertisedListenerName: proto.String(""),
158160
},
159161
},
160162
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -165,7 +167,7 @@ func TestTlsLookupSuccess(t *testing.T) {
165167
BrokerServiceUrlTls: proto.String("pulsar+ssl://broker-1:6651"),
166168
},
167169
},
168-
}, url, serviceNameResolver, true, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
170+
}, url, serviceNameResolver, true, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
169171

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

186188
expectedRequests: []pb.CommandLookupTopic{
187189
{
188-
RequestId: proto.Uint64(1),
189-
Topic: proto.String("my-topic"),
190-
Authoritative: proto.Bool(false),
190+
RequestId: proto.Uint64(1),
191+
Topic: proto.String("my-topic"),
192+
Authoritative: proto.Bool(false),
193+
AdvertisedListenerName: proto.String(""),
191194
},
192195
},
193196
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -199,7 +202,7 @@ func TestLookupWithProxy(t *testing.T) {
199202
ProxyThroughServiceUrl: proto.Bool(true),
200203
},
201204
},
202-
}, url, serviceNameResolver, false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
205+
}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
203206

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

219222
expectedRequests: []pb.CommandLookupTopic{
220223
{
221-
RequestId: proto.Uint64(1),
222-
Topic: proto.String("my-topic"),
223-
Authoritative: proto.Bool(false),
224+
RequestId: proto.Uint64(1),
225+
Topic: proto.String("my-topic"),
226+
Authoritative: proto.Bool(false),
227+
AdvertisedListenerName: proto.String(""),
224228
},
225229
},
226230
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -232,7 +236,7 @@ func TestTlsLookupWithProxy(t *testing.T) {
232236
ProxyThroughServiceUrl: proto.Bool(true),
233237
},
234238
},
235-
}, url, NewPulsarServiceNameResolver(url), true, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
239+
}, url, NewPulsarServiceNameResolver(url), true, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
236240

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

253257
expectedRequests: []pb.CommandLookupTopic{
254258
{
255-
RequestId: proto.Uint64(1),
256-
Topic: proto.String("my-topic"),
257-
Authoritative: proto.Bool(false),
259+
RequestId: proto.Uint64(1),
260+
Topic: proto.String("my-topic"),
261+
Authoritative: proto.Bool(false),
262+
AdvertisedListenerName: proto.String(""),
258263
},
259264
{
260-
RequestId: proto.Uint64(2),
261-
Topic: proto.String("my-topic"),
262-
Authoritative: proto.Bool(true),
265+
RequestId: proto.Uint64(2),
266+
Topic: proto.String("my-topic"),
267+
Authoritative: proto.Bool(true),
268+
AdvertisedListenerName: proto.String(""),
263269
},
264270
},
265271
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -276,7 +282,7 @@ func TestLookupWithRedirect(t *testing.T) {
276282
BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
277283
},
278284
},
279-
}, url, NewPulsarServiceNameResolver(url), false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
285+
}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
280286

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

297303
expectedRequests: []pb.CommandLookupTopic{
298304
{
299-
RequestId: proto.Uint64(1),
300-
Topic: proto.String("my-topic"),
301-
Authoritative: proto.Bool(false),
305+
RequestId: proto.Uint64(1),
306+
Topic: proto.String("my-topic"),
307+
Authoritative: proto.Bool(false),
308+
AdvertisedListenerName: proto.String(""),
302309
},
303310
{
304-
RequestId: proto.Uint64(2),
305-
Topic: proto.String("my-topic"),
306-
Authoritative: proto.Bool(true),
311+
RequestId: proto.Uint64(2),
312+
Topic: proto.String("my-topic"),
313+
Authoritative: proto.Bool(true),
314+
AdvertisedListenerName: proto.String(""),
307315
},
308316
},
309317
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -320,7 +328,7 @@ func TestTlsLookupWithRedirect(t *testing.T) {
320328
BrokerServiceUrlTls: proto.String("pulsar+ssl://broker-1:6651"),
321329
},
322330
},
323-
}, url, NewPulsarServiceNameResolver(url), true, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
331+
}, url, NewPulsarServiceNameResolver(url), true, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
324332

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

340348
expectedRequests: []pb.CommandLookupTopic{
341349
{
342-
RequestId: proto.Uint64(1),
343-
Topic: proto.String("my-topic"),
344-
Authoritative: proto.Bool(false),
350+
RequestId: proto.Uint64(1),
351+
Topic: proto.String("my-topic"),
352+
Authoritative: proto.Bool(false),
353+
AdvertisedListenerName: proto.String(""),
345354
},
346355
},
347356
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -353,7 +362,7 @@ func TestLookupWithInvalidUrlResponse(t *testing.T) {
353362
ProxyThroughServiceUrl: proto.Bool(false),
354363
},
355364
},
356-
}, url, NewPulsarServiceNameResolver(url), false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
365+
}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
357366

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

370379
expectedRequests: []pb.CommandLookupTopic{
371380
{
372-
RequestId: proto.Uint64(1),
373-
Topic: proto.String("my-topic"),
374-
Authoritative: proto.Bool(false),
381+
RequestId: proto.Uint64(1),
382+
Topic: proto.String("my-topic"),
383+
Authoritative: proto.Bool(false),
384+
AdvertisedListenerName: proto.String(""),
375385
},
376386
},
377387
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -381,7 +391,7 @@ func TestLookupWithLookupFailure(t *testing.T) {
381391
Authoritative: proto.Bool(true),
382392
},
383393
},
384-
}, url, NewPulsarServiceNameResolver(url), false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
394+
}, url, NewPulsarServiceNameResolver(url), false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
385395

386396
lr, err := ls.Lookup("my-topic")
387397
assert.Error(t, err)
@@ -468,7 +478,7 @@ func TestGetPartitionedTopicMetadataSuccess(t *testing.T) {
468478
Response: pb.CommandPartitionedTopicMetadataResponse_Success.Enum(),
469479
},
470480
},
471-
}, url, serviceNameResolver, false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
481+
}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
472482

473483
metadata, err := ls.GetPartitionedTopicMetadata("my-topic")
474484
assert.NoError(t, err)
@@ -486,9 +496,10 @@ func TestLookupSuccessWithMultipleHosts(t *testing.T) {
486496

487497
expectedRequests: []pb.CommandLookupTopic{
488498
{
489-
RequestId: proto.Uint64(1),
490-
Topic: proto.String("my-topic"),
491-
Authoritative: proto.Bool(false),
499+
RequestId: proto.Uint64(1),
500+
Topic: proto.String("my-topic"),
501+
AdvertisedListenerName: proto.String(""),
502+
Authoritative: proto.Bool(false),
492503
},
493504
},
494505
mockedResponses: []pb.CommandLookupTopicResponse{
@@ -499,7 +510,7 @@ func TestLookupSuccessWithMultipleHosts(t *testing.T) {
499510
BrokerServiceUrl: proto.String("pulsar://broker-1:6650"),
500511
},
501512
},
502-
}, url, serviceNameResolver, false, log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
513+
}, url, serviceNameResolver, false, "", log.DefaultNopLogger(), NewMetricsProvider(map[string]string{}))
503514

504515
lr, err := ls.Lookup("my-topic")
505516
assert.NoError(t, err)

0 commit comments

Comments
 (0)