Skip to content

Commit

Permalink
remove functional interfaces about dealing with messaging contexts (#…
Browse files Browse the repository at this point in the history 
…27288)
  • Loading branch information
@yiliuTo
yiliuTo authored Feb 25, 2022
1 parent 2c115d8 commit b221786
Show file tree
Hide file tree
Showing 12 changed files with 22 additions and 98 deletions.
9 changes: 3 additions & 6 deletions ...e/src/main/java/com/azure/spring/service/eventhubs/processor/EventProcessingListener.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,6 @@
import com.azure.messaging.eventhubs.models.CloseContext;
import com.azure.messaging.eventhubs.models.ErrorContext;
import com.azure.messaging.eventhubs.models.InitializationContext;
import com.azure.spring.service.eventhubs.processor.consumer.EventHubsCloseContextConsumer;
import com.azure.spring.service.eventhubs.processor.consumer.EventHubsErrorContextConsumer;
import com.azure.spring.service.eventhubs.processor.consumer.EventHubsInitializationContextConsumer;

import java.util.function.Consumer;

Expand All @@ -22,23 +19,23 @@ public interface EventProcessingListener {
* Return the {@link Consumer} of {@link InitializationContext} for Event Hubs by default.
* @return the initialization context consumer.
*/
default EventHubsInitializationContextConsumer getInitializationContextConsumer() {
default Consumer<InitializationContext> getInitializationContextConsumer() {
return initializationContextConsumer -> { };
}

/**
* Return the {@link Consumer} of {@link CloseContext} for Event Hubs by default.
* @return the close context consumer.
*/
default EventHubsCloseContextConsumer getCloseContextConsumer() {
default Consumer<CloseContext> getCloseContextConsumer() {
return closeContext -> { };
}

/**
* Return the {@link Consumer} of {@link ErrorContext} for Event Hubs by default.
* @return the error context consumer.
*/
default EventHubsErrorContextConsumer getErrorContextConsumer() {
default Consumer<ErrorContext> getErrorContextConsumer() {
return errorContext -> { };
}

Expand Down
15 changes: 0 additions & 15 deletions .../com/azure/spring/service/eventhubs/processor/consumer/EventHubsCloseContextConsumer.java
View file

This file was deleted.

15 changes: 0 additions & 15 deletions .../com/azure/spring/service/eventhubs/processor/consumer/EventHubsErrorContextConsumer.java
View file

This file was deleted.

15 changes: 0 additions & 15 deletions ...e/spring/service/eventhubs/processor/consumer/EventHubsInitializationContextConsumer.java
View file

This file was deleted.

7 changes: 0 additions & 7 deletions ...ice/src/main/java/com/azure/spring/service/eventhubs/processor/consumer/package-info.java
View file

This file was deleted.

3 changes: 1 addition & 2 deletions ...rc/main/java/com/azure/spring/service/servicebus/processor/MessageProcessingListener.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,6 @@
package com.azure.spring.service.servicebus.processor;

import com.azure.messaging.servicebus.ServiceBusErrorContext;
import com.azure.spring.service.servicebus.processor.consumer.ServiceBusErrorContextConsumer;

import java.util.function.Consumer;

Expand All @@ -17,7 +16,7 @@ public interface MessageProcessingListener {
* Return the {@link Consumer} of {@link ServiceBusErrorContext} for Event Hubs by default.
* @return the error context consumer.
*/
default ServiceBusErrorContextConsumer getErrorContextConsumer() {
default Consumer<ServiceBusErrorContext> getErrorContextConsumer() {
return errorContext -> { };
}

Expand Down
15 changes: 0 additions & 15 deletions ...om/azure/spring/service/servicebus/processor/consumer/ServiceBusErrorContextConsumer.java
View file

This file was deleted.

7 changes: 0 additions & 7 deletions ...ce/src/main/java/com/azure/spring/service/servicebus/processor/consumer/package-info.java
View file

This file was deleted.

6 changes: 3 additions & 3 deletions ...st/java/com/azure/spring/cloud/stream/binder/eventhubs/EventHubsHealthIndicatorTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import com.azure.spring.service.eventhubs.processor.BatchEventProcessingListener;
import com.azure.spring.service.eventhubs.processor.EventProcessingListener;
import com.azure.spring.service.eventhubs.processor.RecordEventProcessingListener;
import com.azure.spring.service.eventhubs.processor.consumer.EventHubsErrorContextConsumer;
import com.azure.spring.service.eventhubs.properties.EventBatchProperties;
import com.azure.storage.blob.BlobContainerAsyncClient;
import org.junit.jupiter.api.BeforeEach;
Expand All @@ -44,6 +43,7 @@

import java.time.Duration;
import java.util.HashMap;
import java.util.function.Consumer;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.when;
Expand Down Expand Up @@ -263,7 +263,7 @@ private static class TestIntegrationRecordEventProcessingListener implements
private String instrumentationId;

@Override
public EventHubsErrorContextConsumer getErrorContextConsumer() {
public Consumer<ErrorContext> getErrorContextConsumer() {
return errorContext -> {
updateInstrumentation(errorContext, instrumentationManager, instrumentationId);
};
Expand Down Expand Up @@ -292,7 +292,7 @@ private static class TestIntegrationBatchEventProcessingListener implements
private String instrumentationId;

@Override
public EventHubsErrorContextConsumer getErrorContextConsumer() {
public Consumer<ErrorContext> getErrorContextConsumer() {
return errorContext -> {
updateInstrumentation(errorContext, instrumentationManager, instrumentationId);
};
Expand Down
5 changes: 3 additions & 2 deletions .../java/com/azure/spring/cloud/stream/binder/servicebus/ServiceBusHealthIndicatorTests.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

package com.azure.spring.cloud.stream.binder.servicebus;

import com.azure.messaging.servicebus.ServiceBusErrorContext;
import com.azure.messaging.servicebus.ServiceBusReceivedMessageContext;
import com.azure.spring.cloud.stream.binder.servicebus.properties.ServiceBusBindingProperties;
import com.azure.spring.cloud.stream.binder.servicebus.properties.ServiceBusConsumerProperties;
Expand All @@ -15,7 +16,6 @@
import com.azure.spring.messaging.AzureHeaders;
import com.azure.spring.messaging.checkpoint.CheckpointMode;
import com.azure.spring.service.servicebus.processor.RecordMessageProcessingListener;
import com.azure.spring.service.servicebus.processor.consumer.ServiceBusErrorContextConsumer;
import com.azure.spring.service.servicebus.properties.ServiceBusEntityType;
import com.azure.spring.servicebus.core.ServiceBusProcessorContainer;
import com.azure.spring.servicebus.core.ServiceBusTemplate;
Expand All @@ -39,6 +39,7 @@
import org.springframework.test.util.ReflectionTestUtils;

import java.util.HashMap;
import java.util.function.Consumer;

import static com.azure.spring.integration.instrumentation.Instrumentation.Type.CONSUMER;
import static org.assertj.core.api.Assertions.assertThat;
Expand Down Expand Up @@ -201,7 +202,7 @@ public void onMessage(ServiceBusReceivedMessageContext messageContext) {
}

@Override
public ServiceBusErrorContextConsumer getErrorContextConsumer() {
public Consumer<ServiceBusErrorContext> getErrorContextConsumer() {
return errorContext -> {
if (instrumentation != null) {
if (instrumentation instanceof ServiceBusProcessorInstrumentation) {
Expand Down
18 changes: 9 additions & 9 deletions ...n/java/com/azure/spring/integration/eventhubs/inbound/EventHubsInboundChannelAdapter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,11 @@
package com.azure.spring.integration.eventhubs.inbound;

import com.azure.messaging.eventhubs.EventData;
import com.azure.messaging.eventhubs.models.CloseContext;
import com.azure.messaging.eventhubs.models.ErrorContext;
import com.azure.messaging.eventhubs.models.EventBatchContext;
import com.azure.messaging.eventhubs.models.EventContext;
import com.azure.messaging.eventhubs.models.InitializationContext;
import com.azure.messaging.eventhubs.models.PartitionContext;
import com.azure.spring.eventhubs.checkpoint.CheckpointManagers;
import com.azure.spring.eventhubs.checkpoint.EventCheckpointManager;
Expand All @@ -27,9 +29,6 @@
import com.azure.spring.service.eventhubs.processor.BatchEventProcessingListener;
import com.azure.spring.service.eventhubs.processor.EventProcessingListener;
import com.azure.spring.service.eventhubs.processor.RecordEventProcessingListener;
import com.azure.spring.service.eventhubs.processor.consumer.EventHubsCloseContextConsumer;
import com.azure.spring.service.eventhubs.processor.consumer.EventHubsErrorContextConsumer;
import com.azure.spring.service.eventhubs.processor.consumer.EventHubsInitializationContextConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.integration.endpoint.MessageProducerSupport;
Expand All @@ -39,6 +38,7 @@

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

/**
* Inbound channel adapter for Azure Event Hubs.
Expand Down Expand Up @@ -240,7 +240,7 @@ private class IntegrationRecordEventProcessingListener implements Instrumentatio
private String instrumentationId;

@Override
public EventHubsErrorContextConsumer getErrorContextConsumer() {
public Consumer<ErrorContext> getErrorContextConsumer() {
return errorContext -> {
LOGGER.error("Record event error occurred on partition: {}. Error: {}",
errorContext.getPartitionContext().getPartitionId(),
Expand Down Expand Up @@ -273,14 +273,14 @@ public void onEvent(EventContext eventContext) {
}

@Override
public EventHubsCloseContextConsumer getCloseContextConsumer() {
public Consumer<CloseContext> getCloseContextConsumer() {
return closeContext -> LOGGER.info("Stopped receiving on partition: {}. Reason: {}",
closeContext.getPartitionContext().getPartitionId(),
closeContext.getCloseReason());
}

@Override
public EventHubsInitializationContextConsumer getInitializationContextConsumer() {
public Consumer<InitializationContext> getInitializationContextConsumer() {
return initializationContext -> LOGGER.info("Started receiving on partition: {}",
initializationContext.getPartitionContext().getPartitionId());
}
Expand Down Expand Up @@ -323,7 +323,7 @@ private class IntegrationBatchEventProcessingListener implements Instrumentation
private String instrumentationId;

@Override
public EventHubsErrorContextConsumer getErrorContextConsumer() {
public Consumer<ErrorContext> getErrorContextConsumer() {
return errorContext -> {
LOGGER.error("Error occurred on partition: {}. Error: {}",
errorContext.getPartitionContext().getPartitionId(),
Expand All @@ -333,14 +333,14 @@ public EventHubsErrorContextConsumer getErrorContextConsumer() {
}

@Override
public EventHubsCloseContextConsumer getCloseContextConsumer() {
public Consumer<CloseContext> getCloseContextConsumer() {
return closeContext -> LOGGER.info("Stopped receiving on partition: {}. Reason: {}",
closeContext.getPartitionContext().getPartitionId(),
closeContext.getCloseReason());
}

@Override
public EventHubsInitializationContextConsumer getInitializationContextConsumer() {
public Consumer<InitializationContext> getInitializationContextConsumer() {
return initializationContext -> LOGGER.info("Started receiving on partition: {}",
initializationContext.getPartitionContext().getPartitionId());
}
Expand Down
5 changes: 3 additions & 2 deletions ...java/com/azure/spring/integration/servicebus/inbound/ServiceBusInboundChannelAdapter.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

package com.azure.spring.integration.servicebus.inbound;

import com.azure.messaging.servicebus.ServiceBusErrorContext;
import com.azure.messaging.servicebus.ServiceBusMessage;
import com.azure.messaging.servicebus.ServiceBusReceivedMessage;
import com.azure.messaging.servicebus.ServiceBusReceivedMessageContext;
Expand All @@ -18,7 +19,6 @@
import com.azure.spring.messaging.converter.AzureMessageConverter;
import com.azure.spring.service.servicebus.processor.MessageProcessingListener;
import com.azure.spring.service.servicebus.processor.RecordMessageProcessingListener;
import com.azure.spring.service.servicebus.processor.consumer.ServiceBusErrorContextConsumer;
import com.azure.spring.service.servicebus.properties.ServiceBusEntityType;
import com.azure.spring.servicebus.core.ServiceBusProcessorContainer;
import com.azure.spring.servicebus.support.converter.ServiceBusMessageConverter;
Expand All @@ -34,6 +34,7 @@

import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

import static com.azure.spring.service.servicebus.properties.ServiceBusEntityType.QUEUE;
import static com.azure.spring.service.servicebus.properties.ServiceBusEntityType.TOPIC;
Expand Down Expand Up @@ -182,7 +183,7 @@ protected class IntegrationRecordMessageProcessingListener implements RecordMess
private String instrumentationId;

@Override
public ServiceBusErrorContextConsumer getErrorContextConsumer() {
public Consumer<ServiceBusErrorContext> getErrorContextConsumer() {
return errorContext -> {
LOGGER.error("Error occurred on entity {}. Error: {}",
errorContext.getEntityPath(),
Expand Down

0 comments on commit b221786

Please sign in to comment.