Add support for RPC

With RPC client and server API.

Author: acogoluegnes

src/main/java/com/rabbitmq/model/Connection.java

@@ -27,6 +27,10 @@ public interface Connection extends Closeable, Resource {
 
   ConsumerBuilder consumerBuilder();
 
+  RpcClientBuilder rpcClientBuilder();
+
+  RpcServerBuilder rpcServerBuilder();
+
   @Override
   void close();
 }

src/main/java/com/rabbitmq/model/Message.java

@@ -37,6 +37,16 @@ public interface Message {
 
   UUID messageIdAsUuid();
 
+  Object correlationId();
+
+  String correlationIdAsString();
+
+  long correlationIdAsLong();
+
+  byte[] correlationIdAsBinary();
+
+  UUID correlationIdAsUuid();
+
   byte[] userId();
 
   String to();
@@ -45,6 +55,8 @@ public interface Message {
 
   String replyTo();
 
+  Message messageId(Object id);
+
   Message messageId(String id);
 
   Message messageId(long id);
@@ -53,6 +65,16 @@ public interface Message {
 
   Message messageId(UUID id);
 
+  Message correlationId(Object correlationId);
+
+  Message correlationId(String correlationId);
+
+  Message correlationId(long correlationId);
+
+  Message correlationId(byte[] correlationId);
+
+  Message correlationId(UUID correlationId);
+
   Message userId(byte[] userId);
 
   Message to(String address);

src/main/java/com/rabbitmq/model/RpcClient.java

@@ -0,0 +1,32 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// 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.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.model;
+
+import java.util.concurrent.CompletableFuture;
+
+public interface RpcClient extends AutoCloseable {
+
+  Message message();
+
+  Message message(byte[] body);
+
+  CompletableFuture<Message> publish(Message message);
+
+  @Override
+  void close();
+}

src/main/java/com/rabbitmq/model/RpcClientBuilder.java

@@ -0,0 +1,32 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// 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.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.model;
+
+public interface RpcClientBuilder {
+
+  RpcClientAddressBuilder requestAddress();
+
+  RpcClientBuilder replyToQueue(String replyToQueue);
+
+  RpcClient build();
+
+  interface RpcClientAddressBuilder extends AddressBuilder<RpcClientAddressBuilder> {
+
+    RpcClientBuilder rpcClient();
+  }
+}

src/main/java/com/rabbitmq/model/RpcServer.java

@@ -0,0 +1,36 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// 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.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.model;
+
+public interface RpcServer extends AutoCloseable {
+
+  interface Handler {
+
+    Message handle(Context ctx, Message request);
+  }
+
+  interface Context {
+
+    Message message();
+
+    Message message(byte[] body);
+  }
+
+  @Override
+  void close();
+}

src/main/java/com/rabbitmq/model/RpcServerBuilder.java

@@ -0,0 +1,34 @@
+// Copyright (c) 2024 Broadcom. All Rights Reserved.
+// The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+//
+// 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.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+package com.rabbitmq.model;
+
+public interface RpcServerBuilder {
+
+  RpcServerBuilder requestQueue(String requestQueue);
+
+  RpcServerBuilder handler(RpcServer.Handler handler);
+
+  RpcServerAddressBuilder replyToAddress();
+
+  RpcServer build();
+
+  interface RpcServerAddressBuilder extends AddressBuilder<RpcServerAddressBuilder> {
+
+    RpcServerBuilder rpcServer();
+  }
+}

src/main/java/com/rabbitmq/model/amqp/AmqpConnection.java

@@ -152,6 +152,16 @@ public ConsumerBuilder consumerBuilder() {
     return new AmqpConsumerBuilder(this);
   }
 
+  @Override
+  public RpcClientBuilder rpcClientBuilder() {
+    return new RpcSupport.AmqpRpcClientBuilder(this);
+  }
+
+  @Override
+  public RpcServerBuilder rpcServerBuilder() {
+    return new RpcSupport.AmqpRpcServerBuilder(this);
+  }
+
   @Override
   public void close() {
     if (this.closed.compareAndSet(false, true)) {

src/main/java/com/rabbitmq/model/amqp/AmqpMessage.java

@@ -87,6 +87,31 @@ public UUID messageIdAsUuid() {
     return returnFromDelegate(m -> (UUID) m.messageId());
   }
 
+  @Override
+  public Object correlationId() {
+    return returnFromDelegate(org.apache.qpid.protonj2.client.Message::correlationId);
+  }
+
+  @Override
+  public String correlationIdAsString() {
+    return returnFromDelegate(m -> (String) m.correlationId());
+  }
+
+  @Override
+  public long correlationIdAsLong() {
+    return returnFromDelegate(m -> ((UnsignedLong) m.correlationId())).longValue();
+  }
+
+  @Override
+  public byte[] correlationIdAsBinary() {
+    return returnFromDelegate(m -> ((Binary) m.correlationId()).asByteArray());
+  }
+
+  @Override
+  public UUID correlationIdAsUuid() {
+    return returnFromDelegate(m -> (UUID) m.correlationId());
+  }
+
   @Override
   public byte[] userId() {
     return returnFromDelegate(org.apache.qpid.protonj2.client.Message::userId);
@@ -107,6 +132,12 @@ public String replyTo() {
     return returnFromDelegate(org.apache.qpid.protonj2.client.Message::replyTo);
   }
 
+  @Override
+  public Message messageId(Object id) {
+    callOnDelegate(m -> m.messageId(id));
+    return this;
+  }
+
   @Override
   public Message messageId(String id) {
     callOnDelegate(m -> m.messageId(id));
@@ -131,6 +162,36 @@ public Message messageId(UUID id) {
     return this;
   }
 
+  @Override
+  public Message correlationId(Object correlationId) {
+    callOnDelegate(m -> m.correlationId(correlationId));
+    return this;
+  }
+
+  @Override
+  public Message correlationId(String correlationId) {
+    callOnDelegate(m -> m.correlationId(correlationId));
+    return this;
+  }
+
+  @Override
+  public Message correlationId(long correlationId) {
+    callOnDelegate(m -> m.correlationId(correlationId));
+    return this;
+  }
+
+  @Override
+  public Message correlationId(byte[] correlationId) {
+    callOnDelegate(m -> m.correlationId(correlationId));
+    return this;
+  }
+
+  @Override
+  public Message correlationId(UUID correlationId) {
+    callOnDelegate(m -> m.correlationId(correlationId));
+    return this;
+  }
+
   @Override
   public Message userId(byte[] userId) {
     callOnDelegate(m -> m.userId(userId));
@@ -297,10 +358,10 @@ public Object removeProperty(String key) {
 
   @Override
   public MessageAddressBuilder address() {
-    return new DefaultMessageAddressBuilder<>(this);
+    return new DefaultMessageAddressBuilder(this);
   }
 
-  private static class DefaultMessageAddressBuilder<T>
+  private static class DefaultMessageAddressBuilder
       extends DefaultAddressBuilder<MessageAddressBuilder> implements MessageAddressBuilder {
 
     private final Message message;

src/main/java/com/rabbitmq/model/amqp/AmqpPublisherBuilder.java

@@ -17,6 +17,7 @@
 // [email protected].
 package com.rabbitmq.model.amqp;
 
+import com.rabbitmq.model.AddressBuilder;
 import com.rabbitmq.model.Publisher;
 import com.rabbitmq.model.PublisherBuilder;
 import com.rabbitmq.model.Resource;
@@ -72,6 +73,10 @@ List<Resource.StateListener> listeners() {
     return listeners;
   }
 
+  AddressBuilder<?> addressBuilder() {
+    return this.addressBuilder;
+  }
+
   String address() {
     return this.addressBuilder.address();
   }