Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
AXON-11603 Rewrite distributed commandbus to have a separate service …
…registry, Added websockets implementation
- Loading branch information
1 parent
f8489b8
commit 0ee3e51
Showing
45 changed files
with
1,934 additions
and
1,583 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
25 changes: 25 additions & 0 deletions
25
.../axonframework/commandhandling/distributed/CommandBusConnectorCommunicationException.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| package org.axonframework.commandhandling.distributed; | ||
|
|
||
| import org.axonframework.common.AxonTransientException; | ||
|
|
||
| /** | ||
| * Exception thrown when the CommandBusConnector has a communication failure | ||
| */ | ||
| public class CommandBusConnectorCommunicationException extends AxonTransientException { | ||
| /** | ||
| * Initializes the CommandBusConnectorCommunicationException | ||
| * @param message The message of the exception | ||
| */ | ||
| public CommandBusConnectorCommunicationException(String message) { | ||
| super(message); | ||
| } | ||
|
|
||
| /** | ||
| * Initializes the CommandBusConnectorCommunicationException | ||
| * @param message The message of the exception | ||
| * @param cause The cause of this exception | ||
| */ | ||
| public CommandBusConnectorCommunicationException(String message, Throwable cause) { | ||
| super(message, cause); | ||
| } | ||
| } |
41 changes: 41 additions & 0 deletions
41
...rc/main/java/org/axonframework/commandhandling/distributed/CommandCallbackRepository.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| package org.axonframework.commandhandling.distributed; | ||
|
|
||
| import java.util.Iterator; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| /** | ||
| * @author koen | ||
| * on 11-5-16. | ||
| */ | ||
| public class CommandCallbackRepository<A> { | ||
| final Map<String, CommandCallbackWrapper> callbacks = new ConcurrentHashMap<>(); | ||
|
|
||
| public void cancelCallbacks(A channelId) { | ||
| Iterator<CommandCallbackWrapper> callbacks = this.callbacks.values().iterator(); | ||
| while (callbacks.hasNext()) { | ||
| CommandCallbackWrapper wrapper = callbacks.next(); | ||
| if (wrapper.getChannelIdentifier().equals(channelId)) { | ||
| wrapper.fail(new CommandBusConnectorCommunicationException(String.format( | ||
| "Connection error while waiting for a response on command %s", | ||
| wrapper.getMessage().getCommandName()))); | ||
| callbacks.remove(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| public <A, C, R> CommandCallbackWrapper<A, C, R> fetchAndRemove(String callbackId) { | ||
| return callbacks.remove(callbackId); | ||
| } | ||
|
|
||
| public <A, C, R> void store(String callbackId, CommandCallbackWrapper<A, C, R> commandCallbackWrapper) { | ||
| CommandCallbackWrapper previous; | ||
| if ((previous = callbacks.put(callbackId, commandCallbackWrapper)) != null) { | ||
| //a previous callback with the same command ID was already found, we will cancel the callback as the command | ||
| //is likely to be retried, so the previous one likely failed | ||
| previous.fail(new CommandBusConnectorCommunicationException( | ||
| "Command-callback cancelled, a new command with the same ID is entered into the command bus")); | ||
| } | ||
| } | ||
| } |
47 changes: 47 additions & 0 deletions
47
...s/src/main/java/org/axonframework/commandhandling/distributed/CommandCallbackWrapper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| package org.axonframework.commandhandling.distributed; | ||
|
|
||
| import org.axonframework.commandhandling.CommandCallback; | ||
| import org.axonframework.commandhandling.CommandMessage; | ||
|
|
||
| /** | ||
| * @author koen | ||
| * on 11-5-16. | ||
| */ | ||
|
|
||
| public class CommandCallbackWrapper<A, C, R> implements CommandCallback<C, R> { | ||
| private final CommandCallback<? super C, R> wrapped; | ||
| private final A sessionId; | ||
| private final CommandMessage<C> message; | ||
|
|
||
| public CommandCallbackWrapper(A sessionId, CommandMessage<C> message, CommandCallback<? super C, R> callback) { | ||
| this.wrapped = callback; | ||
| this.sessionId = sessionId; | ||
| this.message = message; | ||
| } | ||
|
|
||
| public CommandMessage<C> getMessage() { | ||
| return message; | ||
| } | ||
|
|
||
| public A getChannelIdentifier() { | ||
| return sessionId; | ||
| } | ||
|
|
||
| public void fail(Throwable e) { | ||
| onFailure(getMessage(), e); | ||
| } | ||
|
|
||
| public void success(R returnValue) { | ||
| onSuccess(getMessage(), returnValue); | ||
| } | ||
|
|
||
| @Override | ||
| public void onSuccess(CommandMessage<? extends C> message, R result) { | ||
| wrapped.onSuccess(message, result); | ||
| } | ||
|
|
||
| @Override | ||
| public void onFailure(CommandMessage<? extends C> message, Throwable cause) { | ||
| wrapped.onFailure(message, cause); | ||
| } | ||
| } |
Oops, something went wrong.