Skip to content

Commit 102dc1a

Browse files
wolf4oodwolf4ood
authored
refactor: changes registration of dispatchers (#4511)
1 parent 067ded3 commit 102dc1a

File tree

13 files changed

+21
-46
lines changed

13 files changed

+21
-46
lines changed

‎core/common/connector-core/src/main/java/org/eclipse/edc/connector/core/message/RemoteMessageDispatcherRegistryImpl.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,8 @@ public class RemoteMessageDispatcherRegistryImpl implements RemoteMessageDispatc
3333
private final Map<String, RemoteMessageDispatcher> dispatchers = new HashMap<>();
3434

3535
@Override
36-
public void register(RemoteMessageDispatcher dispatcher) {
37-
dispatchers.put(dispatcher.protocol(), dispatcher);
36+
public void register(String protocol, RemoteMessageDispatcher dispatcher) {
37+
dispatchers.put(protocol, dispatcher);
3838
}
3939

4040
@Override

‎core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/contractnegotiation/ContractNegotiationEventDispatchTest.java

+1-2
Original file line numberDiff line numberDiff line change
@@ -101,7 +101,7 @@ void shouldDispatchEventsOnProviderContractNegotiationStateChanges(EventRouter e
101101
ContractDefinitionStore contractDefinitionStore,
102102
PolicyDefinitionStore policyDefinitionStore,
103103
AssetIndex assetIndex) {
104-
dispatcherRegistry.register(succeedingDispatcher());
104+
dispatcherRegistry.register("test", succeedingDispatcher());
105105

106106
when(identityService.verifyJwtToken(eq(tokenRepresentation), isA(VerificationContext.class))).thenReturn(Result.success(token));
107107
eventRouter.register(ContractNegotiationEvent.class, eventSubscriber);
@@ -143,7 +143,6 @@ private ContractRequestMessage createContractOfferRequest(Policy policy, String
143143
@NotNull
144144
private RemoteMessageDispatcher succeedingDispatcher() {
145145
var testDispatcher = mock(RemoteMessageDispatcher.class);
146-
when(testDispatcher.protocol()).thenReturn("test");
147146
when(testDispatcher.dispatch(any(), any())).thenReturn(completedFuture(StatusResult.success("any")));
148147
return testDispatcher;
149148
}

‎core/control-plane/control-plane-aggregate-services/src/test/java/org/eclipse/edc/connector/controlplane/services/transferprocess/TransferProcessEventDispatchTest.java

+4-7
Original file line numberDiff line numberDiff line change
@@ -84,8 +84,6 @@
8484
public class TransferProcessEventDispatchTest {
8585

8686
public static final Duration TIMEOUT = Duration.ofSeconds(30);
87-
private final EventSubscriber eventSubscriber = mock();
88-
8987
@RegisterExtension
9088
static final RuntimeExtension RUNTIME = new RuntimePerClassExtension()
9189
.setConfiguration(Map.of(
@@ -100,6 +98,7 @@ public class TransferProcessEventDispatchTest {
10098
.registerServiceMock(ContractNegotiationStore.class, mock())
10199
.registerServiceMock(ParticipantAgentService.class, mock())
102100
.registerServiceMock(DataPlaneClientFactory.class, mock());
101+
private final EventSubscriber eventSubscriber = mock();
103102

104103
@Test
105104
void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService service,
@@ -128,7 +127,7 @@ void shouldDispatchEventsOnTransferProcessStateChanges(TransferProcessService se
128127

129128
when(agent.getIdentity()).thenReturn(providerId);
130129

131-
dispatcherRegistry.register(getTestDispatcher());
130+
dispatcherRegistry.register("test", getTestDispatcher());
132131
when(policyArchive.findPolicyForContract(matches(transferRequest.getContractId()))).thenReturn(Policy.Builder.newInstance().target("assetId").build());
133132
when(negotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(agreement);
134133
when(agentService.createFor(token)).thenReturn(agent);
@@ -194,7 +193,7 @@ void shouldDispatchEventOnTransferProcessTerminated(TransferProcessService servi
194193
.policy(Policy.Builder.newInstance().build())
195194
.build();
196195
when(negotiationStore.findContractAgreement(transferRequest.getContractId())).thenReturn(agreement);
197-
dispatcherRegistry.register(getTestDispatcher());
196+
dispatcherRegistry.register("test", getTestDispatcher());
198197
eventRouter.register(TransferProcessEvent.class, eventSubscriber);
199198

200199
var initiateResult = service.initiateTransfer(transferRequest);
@@ -213,7 +212,7 @@ void shouldDispatchEventOnTransferProcessTerminated(TransferProcessService servi
213212
@Test
214213
void shouldDispatchEventOnTransferProcessFailure(TransferProcessService service, EventRouter eventRouter, RemoteMessageDispatcherRegistry dispatcherRegistry,
215214
ContractNegotiationStore negotiationStore, PolicyArchive policyArchive) {
216-
dispatcherRegistry.register(getFailingDispatcher());
215+
dispatcherRegistry.register("test", getFailingDispatcher());
217216
eventRouter.register(TransferProcessEvent.class, eventSubscriber);
218217
var transferRequest = createTransferRequest();
219218
var agreement = ContractAgreement.Builder.newInstance()
@@ -233,7 +232,6 @@ void shouldDispatchEventOnTransferProcessFailure(TransferProcessService service,
233232
@NotNull
234233
private RemoteMessageDispatcher getTestDispatcher() {
235234
var testDispatcher = mock(RemoteMessageDispatcher.class);
236-
when(testDispatcher.protocol()).thenReturn("test");
237235
var ack = TransferProcessAck.Builder.newInstance().build();
238236
when(testDispatcher.dispatch(any(), any())).thenReturn(completedFuture(StatusResult.success(ack)));
239237
return testDispatcher;
@@ -242,7 +240,6 @@ private RemoteMessageDispatcher getTestDispatcher() {
242240
@NotNull
243241
private RemoteMessageDispatcher getFailingDispatcher() {
244242
var testDispatcher = mock(RemoteMessageDispatcher.class);
245-
when(testDispatcher.protocol()).thenReturn("test");
246243
when(testDispatcher.dispatch(any(), any())).thenReturn(failedFuture(new EdcException("cannot send message")));
247244
return testDispatcher;
248245
}

‎data-protocols/dsp/dsp-http-core/src/main/java/org/eclipse/edc/protocol/dsp/http/DspHttpCoreExtension.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@
5252
import org.eclipse.edc.transform.spi.TypeTransformerRegistry;
5353
import org.eclipse.edc.validator.spi.JsonObjectValidatorRegistry;
5454

55+
import static org.eclipse.edc.protocol.dsp.http.spi.types.HttpMessageProtocol.DATASPACE_PROTOCOL_HTTP;
5556
import static org.eclipse.edc.protocol.dsp.spi.type.DspConstants.DSP_SCOPE;
5657
import static org.eclipse.edc.spi.constants.CoreConstants.JSON_LD;
5758

@@ -127,7 +128,7 @@ public DspHttpRemoteMessageDispatcher dspHttpRemoteMessageDispatcher(ServiceExte
127128
registerNegotiationPolicyScopes(dispatcher);
128129
registerTransferProcessPolicyScopes(dispatcher);
129130
registerCatalogPolicyScopes(dispatcher);
130-
dispatcherRegistry.register(dispatcher);
131+
dispatcherRegistry.register(DATASPACE_PROTOCOL_HTTP, dispatcher);
131132
return dispatcher;
132133
}
133134

‎data-protocols/dsp/dsp-http-core/src/main/java/org/eclipse/edc/protocol/dsp/http/dispatcher/DspHttpRemoteMessageDispatcherImpl.java

-6
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@
2424
import org.eclipse.edc.protocol.dsp.http.spi.dispatcher.DspHttpRemoteMessageDispatcher;
2525
import org.eclipse.edc.protocol.dsp.http.spi.dispatcher.DspHttpRequestFactory;
2626
import org.eclipse.edc.protocol.dsp.http.spi.dispatcher.response.DspHttpResponseBodyExtractor;
27-
import org.eclipse.edc.protocol.dsp.http.spi.types.HttpMessageProtocol;
2827
import org.eclipse.edc.spi.EdcException;
2928
import org.eclipse.edc.spi.iam.AudienceResolver;
3029
import org.eclipse.edc.spi.iam.IdentityService;
@@ -78,11 +77,6 @@ public DspHttpRemoteMessageDispatcherImpl(EdcHttpClient httpClient,
7877
this.audienceResolver = audienceResolver;
7978
}
8079

81-
@Override
82-
public String protocol() {
83-
return HttpMessageProtocol.DATASPACE_PROTOCOL_HTTP;
84-
}
85-
8680
@Override
8781
public <T, M extends RemoteMessage> CompletableFuture<StatusResult<T>> dispatch(Class<T> responseType, M message) {
8882
var handler = (MessageHandler<M, T>) this.handlers.get(message.getClass());

‎data-protocols/dsp/dsp-http-core/src/test/java/org/eclipse/edc/protocol/dsp/http/dispatcher/DspHttpRemoteMessageDispatcherImplTest.java

-6
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,6 @@
5050
import static java.util.concurrent.CompletableFuture.completedFuture;
5151
import static org.assertj.core.api.Assertions.assertThat;
5252
import static org.eclipse.edc.junit.assertions.AbstractResultAssert.assertThat;
53-
import static org.eclipse.edc.protocol.dsp.http.spi.types.HttpMessageProtocol.DATASPACE_PROTOCOL_HTTP;
5453
import static org.eclipse.edc.spi.response.ResponseStatus.ERROR_RETRY;
5554
import static org.eclipse.edc.spi.response.ResponseStatus.FATAL_ERROR;
5655
import static org.mockito.AdditionalMatchers.and;
@@ -100,11 +99,6 @@ void setUp() {
10099
when(tokenDecorator.decorate(any())).thenAnswer(a -> a.getArgument(0));
101100
}
102101

103-
@Test
104-
void protocol_returnDsp() {
105-
assertThat(dispatcher.protocol()).isEqualTo(DATASPACE_PROTOCOL_HTTP);
106-
}
107-
108102
@Test
109103
void dispatch_noScope() {
110104
var authToken = "token";

‎extensions/control-plane/callback/callback-http-dispatcher/src/main/java/org/eclipse/edc/connector/controlplane/callback/dispatcher/http/CallbackEventDispatcherHttpExtension.java

+4-2
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@
2424
import org.eclipse.edc.spi.system.ServiceExtensionContext;
2525
import org.eclipse.edc.spi.types.TypeManager;
2626

27+
import static org.eclipse.edc.connector.controlplane.callback.dispatcher.http.GenericHttpRemoteDispatcherImpl.CALLBACK_EVENT_HTTP;
28+
2729
@Extension(value = CallbackEventDispatcherHttpExtension.NAME)
2830
public class CallbackEventDispatcherHttpExtension implements ServiceExtension {
2931

@@ -55,13 +57,13 @@ public void initialize(ServiceExtensionContext context) {
5557
var baseDispatcher = new GenericHttpRemoteDispatcherImpl(client);
5658
baseDispatcher.registerDelegate(new CallbackEventRemoteMessageDispatcher(typeManager.getMapper(), vault));
5759

58-
registry.register(baseDispatcher);
60+
registry.register(CALLBACK_EVENT_HTTP, baseDispatcher);
5961
}
6062

6163

6264
private String resolveScheme(String scheme) {
6365
if (scheme.equalsIgnoreCase("https") || scheme.equalsIgnoreCase("http")) {
64-
return GenericHttpRemoteDispatcherImpl.CALLBACK_EVENT_HTTP;
66+
return CALLBACK_EVENT_HTTP;
6567
}
6668
return null;
6769
}

‎extensions/control-plane/callback/callback-http-dispatcher/src/main/java/org/eclipse/edc/connector/controlplane/callback/dispatcher/http/GenericHttpRemoteDispatcherImpl.java

+1-6
Original file line numberDiff line numberDiff line change
@@ -39,16 +39,11 @@ protected GenericHttpRemoteDispatcherImpl(EdcHttpClient httpClient) {
3939
this.httpClient = httpClient;
4040
}
4141

42-
@Override
43-
public String protocol() {
44-
return CALLBACK_EVENT_HTTP;
45-
}
46-
4742
@Override
4843
public <T, M extends RemoteMessage> CompletableFuture<StatusResult<T>> dispatch(Class<T> responseType, M message) {
4944
var delegate = (GenericHttpDispatcherDelegate<M, T>) delegates.get(message.getClass());
5045
if (delegate == null) {
51-
throw new EdcException(format("No %s message dispatcher found for message type %s", protocol(), message.getClass()));
46+
throw new EdcException(format("No %s message dispatcher found for message type %s", CALLBACK_EVENT_HTTP, message.getClass()));
5247
}
5348
var request = delegate.buildRequest(message);
5449
return httpClient.executeAsync(request, emptyList())

‎extensions/control-plane/callback/callback-http-dispatcher/src/test/java/org/eclipse/edc/connector/controlplane/callback/dispatcher/http/GenericHttpRemoteDispatcherWrapperExtensionTest.java

+3-6
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,10 @@
2323
import org.junit.jupiter.api.BeforeEach;
2424
import org.junit.jupiter.api.Test;
2525
import org.junit.jupiter.api.extension.ExtendWith;
26-
import org.mockito.ArgumentMatcher;
2726

2827
import static org.eclipse.edc.connector.controlplane.callback.dispatcher.http.GenericHttpRemoteDispatcherImpl.CALLBACK_EVENT_HTTP;
29-
import static org.mockito.ArgumentMatchers.argThat;
28+
import static org.mockito.ArgumentMatchers.eq;
29+
import static org.mockito.ArgumentMatchers.isA;
3030
import static org.mockito.Mockito.mock;
3131
import static org.mockito.Mockito.verify;
3232

@@ -50,10 +50,7 @@ void setUp(ServiceExtensionContext context, ObjectFactory factory) {
5050
void initialize_shouldRegisterBothDispatcher(ServiceExtensionContext context) {
5151

5252
extension.initialize(context);
53-
verify(registry).register(argThat(dispatcher(CALLBACK_EVENT_HTTP)));
53+
verify(registry).register(eq(CALLBACK_EVENT_HTTP), isA(GenericHttpRemoteDispatcherImpl.class));
5454
}
5555

56-
private ArgumentMatcher<GenericHttpRemoteDispatcherImpl> dispatcher(String scheme) {
57-
return dispatcher -> dispatcher.protocol().equals(scheme);
58-
}
5956
}

‎spi/common/core-spi/src/main/java/org/eclipse/edc/spi/message/RemoteMessageDispatcher.java

+1-5
Original file line numberDiff line numberDiff line change
@@ -24,11 +24,7 @@
2424
*/
2525
public interface RemoteMessageDispatcher {
2626

27-
/**
28-
* Return the protocol this dispatcher uses.
29-
*/
30-
String protocol();
31-
27+
3228
/**
3329
* Binds and sends the message.
3430
*

‎spi/common/core-spi/src/main/java/org/eclipse/edc/spi/message/RemoteMessageDispatcherRegistry.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ public interface RemoteMessageDispatcherRegistry {
3131
/**
3232
* Registers a dispatcher.
3333
*/
34-
void register(RemoteMessageDispatcher dispatcher);
34+
void register(String protocol, RemoteMessageDispatcher dispatcher);
3535

3636
/**
3737
* Sends the message.

‎spi/control-plane/control-plane-spi/src/main/java/org/eclipse/edc/connector/controlplane/services/spi/callback/CallbackProtocolResolver.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919

2020
/**
2121
* The resolver translate the scheme part {@link CallbackAddress#getUri()} to an internal
22-
* naming of {@link RemoteMessageDispatcher#protocol()} ()}
22+
* naming of {@link RemoteMessageDispatcher}
2323
*/
2424
@FunctionalInterface
2525
public interface CallbackProtocolResolver {

‎spi/control-plane/control-plane-spi/src/main/java/org/eclipse/edc/connector/controlplane/services/spi/callback/CallbackProtocolResolverRegistry.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020

2121
/**
2222
* Registry for {@link CallbackProtocolResolver} resolvers. The registry resolves the scheme part {@link CallbackAddress#getUri()} to an internal
23-
* naming of {@link RemoteMessageDispatcher#protocol()}
23+
* naming of {@link RemoteMessageDispatcher}
2424
*/
2525
@ExtensionPoint
2626
public interface CallbackProtocolResolverRegistry {

0 commit comments

Comments
 (0)