InMemorySubscriber Functionality (#135)

The following PR completes the functionality for the client-side of uSubscription service communication meaning to be able to subscribe and unsubscribe to topics. We handle the various subscription states and subscription change notifications now.
I've also closed the gap for code coverage using mockito so we now are at 100% coverage.

#129
#132

.gitignore

@@ -37,6 +37,7 @@ build/
 ### VS Code ###
 .vscode/
 logging/
+.VSCodeCounter
 
 .metals
 .bloop

pom.xml

@@ -325,14 +325,27 @@
                 <version>2.4.2</version>
             </dependency>
 
+            <dependency>
+                <groupId>org.mockito</groupId>
+                <artifactId>mockito-core</artifactId>
+                <version>3.12.4</version>
+                <scope>test</scope>
+            </dependency>
+
+            <!-- Mockito JUnit Jupiter extension -->
+            <dependency>
+                <groupId>org.mockito</groupId>
+                <artifactId>mockito-junit-jupiter</artifactId>
+                <version>3.12.4</version>
+                <scope>test</scope>
+            </dependency>
 
             <dependency>
                 <groupId>com.google.errorprone</groupId>
                 <artifactId>error_prone_annotations</artifactId>
                 <version>2.11.0</version>
             </dependency>
 
-
             <dependency>
                 <groupId>org.checkerframework</groupId>
                 <artifactId>checker-qual</artifactId>
@@ -364,14 +377,6 @@
                 <scope>test</scope>
             </dependency>
 
-            <!--JUnit Jupiter Engine to depend on the JUnit5 engine and JUnit 5 API -->
-            <dependency>
-                <groupId>org.junit.jupiter</groupId>
-                <artifactId>junit-jupiter-engine</artifactId>
-                <version>${junit-jupiter.version}</version>
-                <scope>test</scope>
-            </dependency>
-
         </dependencies>
     </dependencyManagement>
 
@@ -380,14 +385,19 @@
         <dependency>
             <groupId>nl.jqno.equalsverifier</groupId>
             <artifactId>equalsverifier</artifactId>
-            <version>3.14.1</version>
             <scope>test</scope>
         </dependency>
 
-        <!--JUnit Jupiter Engine to depend on the JUnit5 engine and JUnit 5 API -->
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-engine</artifactId>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- Mockito JUnit Jupiter extension -->
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
             <scope>test</scope>
         </dependency>
 

src/main/java/org/eclipse/uprotocol/communication/InMemoryRpcClient.java

@@ -17,9 +17,6 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.CompletionException;
-
 import org.eclipse.uprotocol.transport.UListener;
 import org.eclipse.uprotocol.transport.UTransport;
 import org.eclipse.uprotocol.transport.builder.UMessageBuilder;
@@ -83,54 +80,39 @@ public CompletionStage<UPayload> invokeMethod(UUri methodUri, UPayload requestPa
         UMessageBuilder builder = UMessageBuilder.request(transport.getSource(), methodUri, options.timeout());
         UMessage request;
 
-        try {
-            if (!options.token().isBlank()) {
-                builder.withToken(options.token());
-            }
-            // Build a request uMessage
-            request = builder.build(requestPayload);
-
-            // Create the response future and store it in mRequests
-            CompletableFuture<UMessage> responseFuture = new CompletableFuture<UMessage>()
-                    .orTimeout(request.getAttributes().getTtl(), TimeUnit.MILLISECONDS)
-                    .handle((responseMessage, exception) -> {
-                        if (exception != null) {
-                            if (exception instanceof CompletionException) exception = exception.getCause();
-                            if (exception instanceof TimeoutException) {
-                                throw new UStatusException(UCode.DEADLINE_EXCEEDED, "Request timed out");
-                            } else if (exception instanceof UStatusException) {
-                                throw new UStatusException(((UStatusException) exception).getStatus());
-                            } else {
-                                throw new UStatusException(UCode.UNKNOWN, exception.getMessage());
-                            }
-                        }
-                        return responseMessage;
-                    });
-
-            responseFuture.whenComplete(
-                    (responseMessage, exception) -> mRequests.remove(request.getAttributes().getId()));
-
-            mRequests.compute(request.getAttributes().getId(), (requestId, currentRequest) -> {
-                if (currentRequest != null)
-                    throw new UStatusException(UCode.ALREADY_EXISTS, "Duplicated request found");
-                return responseFuture;
-            });
-
-            // Send the request
-            CompletionStage<UStatus> status = transport.send(request);
-
-            return status.thenApply(s -> {
-                if (s.getCode() != UCode.OK) throw new UStatusException(s);
-                return s;
-            }).thenCompose(s -> responseFuture.thenApply(responseMessage ->
-                    UPayload.pack(responseMessage.getPayload(), responseMessage.getAttributes().getPayloadFormat())
-            ));
-
-        } catch (Exception e) {
-            return CompletableFuture.failedFuture(e);
+        if (!options.token().isBlank()) {
+            builder.withToken(options.token());
         }
+        // Build a request uMessage
+        request = builder.build(requestPayload);
+
+        // Create the response future and store it in mRequests
+        CompletableFuture<UMessage> responseFuture = new CompletableFuture<UMessage>()
+                .orTimeout(request.getAttributes().getTtl(), TimeUnit.MILLISECONDS)
+                .exceptionally(e -> {
+                    throw new UStatusException(UCode.DEADLINE_EXCEEDED, "Request timed out");
+                })
+                .whenComplete((responseMessage, exception) -> mRequests.remove(request.getAttributes().getId()));
+
+        mRequests.compute(request.getAttributes().getId(), (requestId, currentRequest) -> {
+            return responseFuture;
+        });
+
+        // Send the request
+        CompletionStage<UStatus> status = transport.send(request);
+
+        return status.thenApply(s -> {
+            if (s.getCode() != UCode.OK) throw new UStatusException(s);
+            return s;
+        }).thenCompose(s -> responseFuture.thenApply(responseMessage ->
+                UPayload.pack(responseMessage.getPayload(), responseMessage.getAttributes().getPayloadFormat())
+        ));
     }
 
+
+    /**
+     * Close the RPC client and clean up any resources
+     */
     public void close() {
         mRequests.clear();
         transport.unregisterListener(UriFactory.ANY, transport.getSource(), mResponseHandler);

src/main/java/org/eclipse/uprotocol/communication/InMemoryRpcServer.java

@@ -105,12 +105,8 @@ public CompletionStage<UStatus> registerRequestHandler(UUri method, RequestHandl
                 return handler;
             });
             return CompletableFuture.completedFuture(UStatus.newBuilder().setCode(UCode.OK).build());
-        } catch (Exception e) {
-            if (e instanceof UStatusException statusException) {
-                return CompletableFuture.completedFuture(statusException.getStatus());
-            }
-            return CompletableFuture.completedFuture(
-                UStatus.newBuilder().setCode(UCode.INTERNAL).setMessage(e.getMessage()).build());
+        } catch (UStatusException e) {
+            return CompletableFuture.completedFuture(e.getStatus());
         }
     }