Denovo1998 commented on code in PR #6: URL: https://github.com/apache/pulsar-java-contrib/pull/6#discussion_r1750279670
########## pulsar-rpc-contrib/src/test/java/org/apache/pulsar/rpc/contrib/SimpleRpcCallTest.java: ########## @@ -0,0 +1,476 @@ +/* + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.pulsar.rpc.contrib; + +import static java.util.UUID.randomUUID; +import java.io.IOException; +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; +import java.util.function.Function; +import java.util.function.Supplier; +import java.util.regex.Pattern; +import lombok.extern.slf4j.Slf4j; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.api.MessageId; +import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.ProducerBuilder; +import org.apache.pulsar.client.api.PulsarClient; +import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.client.api.TypedMessageBuilder; +import org.apache.pulsar.rpc.contrib.client.PulsarRpcClient; +import org.apache.pulsar.rpc.contrib.client.PulsarRpcClientBuilder; +import org.apache.pulsar.rpc.contrib.client.RequestCallBack; +import org.apache.pulsar.rpc.contrib.server.PulsarRpcServer; +import org.apache.pulsar.rpc.contrib.server.PulsarRpcServerBuilder; +import org.awaitility.Awaitility; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +@Slf4j +public class SimpleRpcCallTest { + private final Supplier<String> correlationIdSupplier = () -> randomUUID().toString(); + private final String topicPrefix = "persistent://public/default/"; + private final String topicBase = "testRpcCall"; + private final String requestTopic = topicBase + "-request"; + private final String replyTopic = topicBase + "-reply"; + Pattern requestTopicPattern = Pattern.compile(topicPrefix + requestTopic + ".*"); + Pattern replyTopicPattern = Pattern.compile(topicPrefix + replyTopic + ".*"); + private final String requestSubBase = requestTopic + "-sub"; + private final String replySubBase = replyTopic + "-sub"; + private final Duration replyTimeout = Duration.ofSeconds(3); + private final String synchronousMessage = "SynchronousRequest"; + private final String asynchronousMessage = "AsynchronousRequest"; + private final Schema<TestRequest> requestSchema = Schema.JSON(TestRequest.class); + private final Schema<TestReply> replySchema = Schema.JSON(TestReply.class); + + private PulsarAdmin pulsarAdmin; + private PulsarClient pulsarClient; + private PulsarRpcClient<TestRequest, TestReply> rpcClient; + private PulsarRpcServer<TestRequest, TestReply> rpcServer; + + private Function<TestRequest, CompletableFuture<TestReply>> requestFunction; + private BiConsumer<String, TestRequest> rollbackFunction; + + @BeforeMethod(alwaysRun = true) + protected void setup() throws Exception { + pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl("http://localhost:8080";).build(); + pulsarAdmin.topics().createPartitionedTopic(requestTopic, 10); + pulsarAdmin.topics().createPartitionedTopic(replyTopic, 10); + pulsarClient = PulsarClient.builder().serviceUrl("pulsar://localhost:6650").build(); + } + + @AfterMethod(alwaysRun = true) + protected void cleanup() throws Exception { + if (rpcServer != null) { + rpcServer.close(); + } + rpcClient.close(); + pulsarClient.close(); + pulsarAdmin.topics().deletePartitionedTopic(requestTopic); + pulsarAdmin.topics().deletePartitionedTopic(replyTopic); + pulsarAdmin.close(); + } + + @Test + public void testRpcCall() throws Exception { + ProducerBuilder<TestRequest> requestProducerBuilder = pulsarClient.newProducer(requestSchema) + .topic(requestTopic) + .enableBatching(false) + .producerName("requestProducer"); + + // 1.Create PulsarRpcClient + rpcClient = createPulsarRpcClient(pulsarClient, requestProducerBuilder, null, null); + + // 2.Create PulsarRpcServer + final int defaultEpoch = 1; + AtomicInteger epoch = new AtomicInteger(defaultEpoch); + // What do we do when we receive the request message + requestFunction = request -> { + epoch.getAndIncrement(); + return CompletableFuture.completedFuture(new TestReply(request.value() + "-----------done")); + }; + // If the server side is stateful, an error occurs after the server side executes 3-1, and a mechanism for + // checking and rolling back needs to be provided. + rollbackFunction = (id, request) -> { + if (epoch.get() != defaultEpoch) { + epoch.set(defaultEpoch); + } + }; + rpcServer = createPulsarRpcServer(pulsarClient, requestSubBase, requestFunction, rollbackFunction, null); + Review Comment: @StevenLuMT The following comment may refactor the code and may have different usage and exception handling. This test case extension needs to be completed after confirmation. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
