Client-side of the Core uP-L3 service UTwin.

The implementation uses the uP-L2 InMemoryRpcClient interface to communicate with the UTwin service.

#160

src/main/java/org/eclipse/uprotocol/client/utwin/v2/InMemoryUTwinClient.java

@@ -0,0 +1,92 @@
+/**
+ * SPDX-FileCopyrightText: 2024 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Apache License Version 2.0 which is available at
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.eclipse.uprotocol.client.utwin.v2;
+
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import org.eclipse.uprotocol.communication.CallOptions;
+import org.eclipse.uprotocol.communication.InMemoryRpcClient;
+import org.eclipse.uprotocol.communication.RpcClient;
+import org.eclipse.uprotocol.communication.RpcMapper;
+import org.eclipse.uprotocol.communication.UPayload;
+import org.eclipse.uprotocol.communication.UStatusException;
+import org.eclipse.uprotocol.core.utwin.v2.GetLastMessagesRequest;
+import org.eclipse.uprotocol.core.utwin.v2.GetLastMessagesResponse;
+import org.eclipse.uprotocol.core.utwin.v2.UTwinProto;
+import org.eclipse.uprotocol.transport.UTransport;
+import org.eclipse.uprotocol.uri.factory.UriFactory;
+import org.eclipse.uprotocol.v1.UCode;
+import org.eclipse.uprotocol.v1.UStatus;
+import org.eclipse.uprotocol.v1.UUri;
+import org.eclipse.uprotocol.v1.UUriBatch;
+
+import com.google.protobuf.Descriptors.ServiceDescriptor;
+
+/**
+ * The uTwin client implementation using InMemory up-L2 communication layer implementation.
+ */
+public class InMemoryUTwinClient implements UTwinClient {
+    private final RpcClient rpcClient;
+
+    private static final ServiceDescriptor UTWIN = UTwinProto.getDescriptor().getServices().get(0);
+
+    // TODO: The following items eventually need to be pulled from generated code
+    private static final UUri GETLASTMESSAGE_METHOD = UriFactory.fromProto(UTWIN, 1);
+
+
+    /**
+     * Create a new instance of the uTwin client using the provided transport.
+     * 
+     * @param transport The transport to use for communication.
+     */
+    public InMemoryUTwinClient(UTransport transport) {
+        this(new InMemoryRpcClient(transport));
+    }
+
+
+    /**
+     * Create a new instance of the uTwin client passing in the RPCClient to use for communication.
+     * 
+     * @param rpcClient The RPC client to use for communication.
+     */
+    public InMemoryUTwinClient(RpcClient rpcClient) {
+        this.rpcClient = rpcClient;
+    }
+
+
+    /**
+     * Fetch the last messages for a batch of topics.
+     * 
+     * @param topics  {@link UUriBatch} batch of 1 or more topics to fetch the last messages for.
+     * @param options The call options.
+     * @return CompletionStage completes successfully with {@link GetLastMessagesResponse} if uTwin was able
+     *         to fetch the topics or completes exceptionally with {@link UStatus} with the failure reason.
+     *         such as {@code UCode.NOT_FOUND}, {@code UCode.PERMISSION_DENIED} etc...
+     */
+    @Override
+    public CompletionStage<GetLastMessagesResponse> getLastMessages(UUriBatch topics, CallOptions options) {
+        Objects.requireNonNull(topics, "topics must not be null");
+        Objects.requireNonNull(options, "options must not be null");
+
+        // Check if topics is empty
+        if (topics.equals(UUriBatch.getDefaultInstance())) {
+            return CompletableFuture.failedFuture(
+                new UStatusException(UCode.INVALID_ARGUMENT, "topics must not be empty"));
+        }
+
+        GetLastMessagesRequest request = GetLastMessagesRequest.newBuilder().setTopics(topics).build();
+        return RpcMapper.mapResponse(rpcClient.invokeMethod(
+            GETLASTMESSAGE_METHOD, UPayload.pack(request), options), GetLastMessagesResponse.class);
+    }   
+}

src/main/java/org/eclipse/uprotocol/client/utwin/v2/UTwinClient.java

@@ -0,0 +1,52 @@
+/**
+ * SPDX-FileCopyrightText: 2024 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Apache License Version 2.0 which is available at
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.eclipse.uprotocol.client.utwin.v2;
+
+import java.util.concurrent.CompletionStage;
+
+import org.eclipse.uprotocol.communication.CallOptions;
+import org.eclipse.uprotocol.core.utwin.v2.GetLastMessagesResponse;
+import org.eclipse.uprotocol.v1.UUriBatch;
+
+/**
+ * The uTwin client-side interface.
+ * 
+ * UTwin is used to fetch the last published message for a given topic. This is the client-side of the 
+ * UTwin Service contract and communicates with a local uTwin service to fetch the last message for a given topic.
+
+ */
+public interface UTwinClient {
+    /**
+     * Fetch the last messages for a batch of topics.
+     * 
+     * @param topics  {@link UUriBatch} batch of 1 or more topics to fetch the last messages for.
+     * @param options The call options.
+     * @return CompletionStage completes successfully with {@link GetLastMessagesResponse} if uTwin was able
+     *         to fetch the topics or completes exceptionally with {@link UStatus} with the failure reason.
+     *         such as {@code UCode.NOT_FOUND}, {@code UCode.PERMISSION_DENIED} etc...
+     */
+    CompletionStage<GetLastMessagesResponse> getLastMessages(UUriBatch topics, CallOptions options);
+
+
+    /**
+     * Fetch the last messages for a batch of topics.
+     * 
+     * @param topics  {@link UUriBatch} batch of 1 or more topics to fetch the last messages for.
+     * @return CompletionStage completes successfully with {@link GetLastMessagesResponse} if uTwin was able
+     *         to fetch the topics or completes exceptionally with {@link UStatus} with the failure reason.
+     *         such as {@code UCode.NOT_FOUND}, {@code UCode.PERMISSION_DENIED} etc...
+     */
+    default CompletionStage<GetLastMessagesResponse> getLastMessages(UUriBatch topics) {
+        return getLastMessages(topics, CallOptions.DEFAULT);
+    }
+}

src/test/java/org/eclipse/uprotocol/client/usubscription/v3/InMemoryUSubscriptionClientTest.java

@@ -18,8 +18,6 @@
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Executors;
 import java.util.concurrent.CyclicBarrier;
-import java.util.concurrent.ExecutionException;
-
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;

src/test/java/org/eclipse/uprotocol/client/utwin/v2/InMemoryUTwinClientTest.java

@@ -0,0 +1,167 @@
+/**
+ * SPDX-FileCopyrightText: 2024 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Apache License Version 2.0 which is available at
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.eclipse.uprotocol.client.utwin.v2;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ExecutionException;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.eclipse.uprotocol.communication.InMemoryRpcClient;
+import org.eclipse.uprotocol.communication.RpcClient;
+import org.eclipse.uprotocol.communication.SimpleNotifier;
+import org.eclipse.uprotocol.communication.UPayload;
+import org.eclipse.uprotocol.communication.UStatusException;
+import org.eclipse.uprotocol.core.utwin.v2.GetLastMessagesResponse;
+import org.eclipse.uprotocol.transport.UTransport;
+import org.eclipse.uprotocol.v1.UCode;
+import org.eclipse.uprotocol.v1.UStatus;
+import org.eclipse.uprotocol.v1.UUri;
+import org.eclipse.uprotocol.v1.UUriBatch;
+
+/**
+ * The uTwin client implementation using InMemory up-L2 communication layer implementation.
+ * This is the test code for said implementation.
+ */
+@ExtendWith(MockitoExtension.class)
+public class InMemoryUTwinClientTest {
+    @Mock
+    private UTransport transport;
+
+
+    private final UUri topic = UUri.newBuilder().setAuthorityName("hartley").setUeId(3)
+        .setUeVersionMajor(1).setResourceId(0x8000).build();
+
+    private final UUri source = UUri.newBuilder().setAuthorityName("Hartley").setUeId(4)
+        .setUeVersionMajor(1).build();
+
+
+    @BeforeEach
+    public void setup() {
+        transport = mock(UTransport.class);
+    }
+
+
+    @Test
+    @DisplayName("Test calling getLastMessages() passing null transport")
+    void testGetLastMessagesNull() {
+        assertThrows(NullPointerException.class, () -> {
+            new InMemoryUTwinClient((UTransport)null);
+        });
+
+    }
+
+
+    @Test
+    @DisplayName("Test calling getLastMessages() passing null topics")
+    void testGetLastMessagesNullTopics() {
+
+        when(transport.registerListener(any(), any(), any())).thenReturn(
+            CompletableFuture.completedFuture(UStatus.newBuilder().setCode(UCode.OK).build()));
+
+        InMemoryUTwinClient client = new InMemoryUTwinClient(transport);
+        assertThrows(NullPointerException.class, () -> {
+            client.getLastMessages(null);
+        });
+    }
+
+
+    @Test
+    @DisplayName("Test calling getLastMessages() passing empty topics")
+    void testGetLastMessagesEmptyTopics() {
+        UUriBatch topics = UUriBatch.getDefaultInstance();
+
+        when(transport.registerListener(any(), any(), any())).thenReturn(
+            CompletableFuture.completedFuture(UStatus.newBuilder().setCode(UCode.OK).build()));
+
+        InMemoryUTwinClient client = new InMemoryUTwinClient(transport);
+        assertThrows(ExecutionException.class, () -> {
+            client.getLastMessages(topics).handle((r, e) -> {
+                assertNotNull(e);
+                UStatusException t = (UStatusException) e.getCause();
+                assertNotNull(t);
+                assertEquals(t.getCode(), UCode.INVALID_ARGUMENT);
+                assertEquals(t.getMessage(), "topics must not be empty");
+                return r;
+            }).toCompletableFuture().get();
+        });
+    }
+
+
+    @Test
+    @DisplayName("Test calling getLastMessages() with valid topics")
+    void testGetLastMessages() {
+
+        RpcClient rpcClient = Mockito.mock(RpcClient.class);
+
+        UUriBatch topics = UUriBatch.newBuilder().addUris(topic).build();
+
+        when(rpcClient.invokeMethod(any(), any(), any())).thenReturn(
+            CompletableFuture.completedFuture(UPayload.pack(GetLastMessagesResponse.getDefaultInstance())));
+
+        InMemoryUTwinClient client = new InMemoryUTwinClient(rpcClient);
+        CompletionStage<GetLastMessagesResponse> response = client.getLastMessages(topics);
+        assertNotNull(response);
+        assertFalse(response.toCompletableFuture().isCompletedExceptionally());
+        assertDoesNotThrow(() -> response.toCompletableFuture().get());
+    }
+
+
+    @Test
+    @DisplayName("Test calling getLastMessages() when the RpcClient completes exceptionally")
+    void testGetLastMessagesException() {
+        RpcClient rpcClient = Mockito.mock(RpcClient.class);
+
+        UUriBatch topics = UUriBatch.newBuilder().addUris(topic).build();
+
+        when(rpcClient.invokeMethod(any(), any(), any())).thenReturn(
+            CompletableFuture.failedFuture(new UStatusException(UCode.NOT_FOUND, "Not found")));
+
+        InMemoryUTwinClient client = new InMemoryUTwinClient(rpcClient);
+        CompletionStage<GetLastMessagesResponse> response = client.getLastMessages(topics);
+        assertNotNull(response);
+        assertTrue(response.toCompletableFuture().isCompletedExceptionally());
+        assertDoesNotThrow(() -> {
+            response
+                .handle((r, e) -> {
+                    assertNotNull(e);
+                    UStatusException t = (UStatusException) e.getCause();
+                    assertNotNull(t);
+                    assertEquals(t.getCode(), UCode.NOT_FOUND);
+                    assertEquals(t.getMessage(), "Not found");
+                    return r;
+                })
+                .toCompletableFuture().get();
+        });
+    }
+
+}