ServiceBusProcessorClients provides a push-based mechanism that invokes the message processing callback when a message is received or the error handler when an error occurs when receiving messages. A ServiceBusProcessorClient can be created to process messages for a session-enabled Service Bus entity or a non session-enabled Service Bus entity.Sample code to start a processor client
Consumer<ServiceBusReceivedMessageContext> messageProcessor = context -> { ServiceBusReceivedMessage message = context.getMessage(); System.out.println("Received message " + message.getBody().toString()); }; Consumer<Throwable> errorHandler = throwable -> { System.out.println("Error when receiving messages " + throwable.getMessage()); if (throwable instanceof ServiceBusReceiverException) { ServiceBusReceiverException serviceBusReceiverException = (ServiceBusReceiverException) throwable; System.out.println("Error source " + serviceBusReceiverException.getErrorSource()); } }; ServiceBusProcessorClient processorClient = new ServiceBusClientBuilder() .connectionString("<< connection-string >>") .processor() .queueName("<< queue name >>") .processMessage(messageProcessor) .processError(errorHandler) .buildProcessorClient(); // Start the processor in the background processorClient.start(); Sample code to start a session-enabled processor client
Consumer<ServiceBusReceivedMessageContext> messageProcessor = context -> { ServiceBusReceivedMessage message = context.getMessage(); System.out.println("Received message " + message.getBody().toString() + " session: " + message.getSessionId()); }; Consumer<Throwable> errorHandler = throwable -> { System.out.println("Error when receiving messages " + throwable.getMessage()); if (throwable instanceof ServiceBusReceiverException) { ServiceBusReceiverException serviceBusReceiverException = (ServiceBusReceiverException) throwable; System.out.println("Error source " + serviceBusReceiverException.getErrorSource()); } }; ServiceBusProcessorClient sessionProcessorClient = new ServiceBusClientBuilder() .connectionString("<< connection-string >>") .sessionProcessor() .queueName("<< session-enabled queue name >>") .maxConcurrentSessions(2) .processMessage(messageProcessor) .processError(errorHandler) .buildProcessorClient(); // Start the processor in the background sessionProcessorClient.start(); ncG1vNJzZmiZqqq%2Fpr%2FDpJuom6Njr627wWeaqKqVY8SqusOorqxmnprBcHDWnploopGrrnCt2a6pnmWdmsC0rcaipaBlo5q%2Ft7XCnpmuq19se3F6j2aZnqyRY4Rwr86mZpqypaeycLnErKqan5mjtHC%2FxKutopuVl8K0e7Keqa%2Bhk5qPtr%2Bvq6acnaOovLOPy6Kcp6xencGuuA%3D%3D