-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
4eb5cfd
commit 9af3744
Showing
7 changed files
with
135 additions
and
88 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
31 changes: 31 additions & 0 deletions
31
rpcnis-core/src/main/java/com/rpcnis/core/trackers/IncomingInvocationTracker.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,31 @@ | ||
| package com.rpcnis.core.trackers; | ||
|
|
||
| import com.rpcnis.core.executor.ImplementationWrapper; | ||
|
|
||
| import java.util.Collection; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| public class IncomingInvocationTracker { | ||
|
|
||
| // Map invocation handlers by type; used for dispatching incoming invocations | ||
| // Handlers without a namespace (which is nullable) are also stored here | ||
| private final ConcurrentHashMap<Class<?>, Collection<ImplementationWrapper>> implementations = new ConcurrentHashMap<>(); | ||
|
|
||
| public void registerImplementation(Object implementation, String namespace) { | ||
| // get the interfaces implemented by the implementation | ||
| Class<?>[] interfaces = implementation.getClass().getInterfaces(); | ||
|
|
||
| // there must be at least one interface | ||
| if (interfaces.length == 0) { | ||
| throw new IllegalArgumentException("Implementation must implement at least one interface/procedure"); | ||
| } | ||
|
|
||
| // register this interface as all the implemented interfaces | ||
| ImplementationWrapper implementationWrapper = new ImplementationWrapper(implementation, namespace); | ||
|
|
||
| for (Class<?> anInterface : interfaces) { | ||
| implementations.computeIfAbsent(anInterface, k -> ConcurrentHashMap.newKeySet()).add(implementationWrapper); | ||
| } | ||
| } | ||
|
|
||
| } |
58 changes: 58 additions & 0 deletions
58
rpcnis-core/src/main/java/com/rpcnis/core/trackers/OutgoingInvocationTracker.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,58 @@ | ||
| package com.rpcnis.core.trackers; | ||
|
|
||
| import com.rpcnis.base.RpcOptions; | ||
| import com.rpcnis.core.models.InvocationDescriptor; | ||
| import com.rpcnis.core.proxy.PendingInvocation; | ||
|
|
||
| import java.util.Timer; | ||
| import java.util.UUID; | ||
| import java.util.concurrent.CompletionException; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| public class OutgoingInvocationTracker { | ||
|
|
||
| private final Timer timer; | ||
| private final RpcOptions options; | ||
|
|
||
| // Map of pending invocations, keyed by invocation id | ||
| private final ConcurrentHashMap<UUID, PendingInvocation<?>> pendingInvocations = new ConcurrentHashMap<>(); | ||
|
|
||
| public OutgoingInvocationTracker(RpcOptions options, Timer timer) { | ||
| this.options = options; | ||
| this.timer = timer; | ||
| } | ||
|
|
||
| public <T> T invokeRemoteMethod(InvocationDescriptor invocationDescriptor) throws Throwable { | ||
| // create a pending invocation | ||
| PendingInvocation<T> pendingInvocation = new PendingInvocation<>(options.getTimeoutSeconds(), this.timer, invocationDescriptor, () -> { | ||
| // remove the pending invocation from the map | ||
| pendingInvocations.remove(invocationDescriptor.getUniqueInvocationId()); | ||
| }); | ||
|
|
||
| // add the pending invocation to the map | ||
| pendingInvocations.put(invocationDescriptor.getUniqueInvocationId(), pendingInvocation); | ||
|
|
||
| // TODO: transmit | ||
|
|
||
| // wait for response or timeout | ||
| try { | ||
| return pendingInvocation.waitForResponse(); | ||
| } catch (CompletionException e) { | ||
| throw e.getCause(); | ||
| } | ||
| } | ||
|
|
||
| public void completeInvocation(InvocationDescriptor invocationDescriptor, Object value) { | ||
| // do we have a pending invocation for this invocation id? | ||
| PendingInvocation<?> pendingInvocation = pendingInvocations.get(invocationDescriptor.getUniqueInvocationId()); | ||
| if (pendingInvocation == null) { | ||
| throw new IllegalStateException("No pending invocation found for invocation id " + invocationDescriptor.getUniqueInvocationId()); | ||
| } | ||
|
|
||
| pendingInvocation.complete(value); | ||
|
|
||
| // remove the pending invocation from the map | ||
| pendingInvocations.remove(invocationDescriptor.getUniqueInvocationId()); | ||
| } | ||
|
|
||
| } |
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
Oops, something went wrong.