Event Hubs API

event-hubs-all-api.md

Publish events

Creating a producer client

// with connection string
EventHubProducerClient producerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string")
                                        .buildProducer();
// with connection string and event hub name                                        
EventHubProducerClient producerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string", "event-hub-name")
                                        .buildProducer();    
// with credential                                        
EventHubProducerClient producerClient = new EventHubClientBuilder()
                                        .credential("fully-qualified-namespace", "event-hub-name", credential)
                                        .buildProducer(); 

// with connection
EventHubConnection eventHubConnection = new EventHubConnection("connection-string");
EventHubProducerClient producerClient = new EventHubClientBuilder()
                                        .connection(eventHubConnection)
                                        .buildProducer();

Sending events

Flux<EventData> events = getEvents();

// send with no partition affinity
producerClient.send(events);

// send with a partition key
SendOptions sendOptions = new SendOptions().partitionKey("my-partition-key");
producerClient.send(events, sendOptions);

// send to a specific partition id
SendOptions sendOptions = new SendOptions().partitionId("0");
producerClient.send(events, sendOptions);

Consume events

Creating a consumer client

// with connection string
EventHubConsumerClient consumerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string")
                                        .buildConsumer();
// with connection string and event hub name
EventHubConsumerClient consumerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string", "event-hub-name")
                                        .buildConsumer();

// with credential
EventHubConsumerClient consumerClient = new EventHubClientBuilder()
                                        .credential("fully-qualified-namespace", "event-hub-name", credential)
                                        .buildConsumer();
// with connection
EventHubConnection eventHubConnection = new EventHubConnection("connection-string");
EventHubConsumerClient consumerClient = new EventHubClientBuilder()
                                        .connection(eventHubConnection)
                                        .buildConsumer();                                        

Consume events

EventHubConsumerClient consumerClient = buildWithOneOfTheMethodsAbove();

// consume events from all partitions
consumerClient.receive().subscribe(partitionEvent -> {
        String partitionId = partitionEvent.getPartitionId();
        EventData eventData = partitionEvent.getEventData();
    });

// consume events from a specific partition
consumerClient.receive(partitionId).subscribe(eventData -> {
        eventData.getBody();
    });    

Consume events with load balancing [Names to be discussed]

// Basic mode
EventHubDistributedConsumerClient distributedConsumerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string")
                                        .checkpointStore(checkpointStore)
                                        .consumerGroup("consumer-group")
                                        .buildDistributedConsumer(); 
                                        
distributedConsumerClient.startProcessing((partitionContext, eventData) -> {/* user code */});

// With partition initialization and close
EventHubDistributedConsumerClient distributedConsumerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string")
                                        .checkpointStore(checkpointStore)
                                        .consumerGroup("consumer-group")
                                        .initializePartition(initializationContext -> {/* user code */})
                                        .closePartition(closeContext -> { /* user code */})
                                        .buildDistributedConsumer();
distributedConsumerClient.startProcessing((partitionContext, eventData) -> {/* user code */});

// With processing error handler
EventHubDistributedConsumerClient distributedConsumerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string")
                                        .checkpointStore(checkpointStore)
                                        .consumerGroup("consumer-group")
                                        .buildDistributedConsumer();
distributedConsumerClient.startProcessing((partitionContext, eventData) -> {/* user code */}, partitionError -> {/* user code */});

// With consumer error handler
EventHubDistributedConsumerClient distributedConsumerClient = new EventHubClientBuilder()
                                        .connectionString("connection-string")
                                        .checkpointStore(checkpointStore)
                                        .consumerGroup("consumer-group")
                                        .consumerError(error -> {/* user code */})
                                        .buildDistributedConsumer();
distributedConsumerClient.startProcessing((partitionContext, eventData) -> {/* user code */}, partitionError -> {/* user code */});

distributedConsumerClient.stopProcessing();