// PulsarClient is an entry point
var client = PulsarClient(
  settings: PulsarClientSettings(
    serviceUrl: 'pulsar://localhost:6650',
  ),
);

// global errors (not associated with individual requests completion)
client.clientErrorStream.listen((Object error) {
});

// create Producer on the topic
var producer = await client.newProducer(
  ProducerCreateParams(topic: 'persistent://tenant/namespace/topic')
);

// send a single message. Result contains the messageId
var resultSingle = await producer.sendMessage(
  GenericMessage(
    propertyMap: {'key': 'value'},
    payload: Uint8List.fromList('Binary payload'.codeUnits),
  )
);

// close individual producer
producer.close();
// close client (with all associated producers, consumers and broker connections)
client.close();

// create Consumer on the topic with defined subscription name and type
var consumer = await client.newConsumer(
  ConsumerCreateParams(
      topic: 'persistent://tenant/namespace/topic',
      subscription: 'subscription-name',
      subType: SubscriptionType.exclusive,
  )
);

// listen for incoming messages
consumer.listen((GenericInputMessage message) {
  // ack message (unless it's intermediate in a batch) 
  if (message.storageType != MessageStorageType.batchIntermediate) {
    consumer.ackMessage(message.messageId, false);
  }
});

// give permission to the broker to push next 100 messages (flow control)
await consumer.getFlow(100);

Future<String> _getAuthToken(String? data) {
  var completer = Completer<String>();

  // alternatively, call refreshCredentials() on existing oauth2 client if supported by the particular grant type 
  oauth2.clientCredentialsGrant(
    oauth2Endpoint,
    oauth2Identifier,
    oauth2Secret,
  )
  .then((oauth2client) {
    completer.complete(oauth2client.credentials.accessToken);
  })
  .catchError((error, trace) { completer.completeError(error, trace); });

  return completer.future;
}

// The authTokenProvider function will be invoked to get/refresh an access token
// The periodic token refresh will be triggered by broker, there's no need to track expiration manually
var client = PulsarClient(
  settings: PulsarClientSettings(
    forceTLS: true,
    serviceUrl: 'pulsar+ssl://broker.my-domain.com:6651',
    authTokenProvider: _getAuthToken,
    authTokenRefreshSupported: true,
  ),
);