Add sendTimeout setting to publisher builder

Author: acogoluegnes

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

@@ -17,9 +17,13 @@
 // [email protected].
 package com.rabbitmq.model;
 
+import java.time.Duration;
+
 public interface PublisherBuilder extends AddressBuilder<PublisherBuilder> {
 
   PublisherBuilder listeners(Resource.StateListener... listeners);
 
+  PublisherBuilder publishTimeout(Duration timeout);
+
   Publisher build();
 }

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

@@ -23,6 +23,8 @@
 import com.rabbitmq.model.ObservationCollector;
 import com.rabbitmq.model.Publisher;
 import com.rabbitmq.model.metrics.MetricsCollector;
+
+import java.time.Duration;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -49,6 +51,7 @@ final class AmqpPublisher extends ResourceBase implements Publisher {
   private final ObservationCollector observationCollector;
   private final Function<Message, Tracker> publishCall;
   private final DefaultAddressBuilder.DestinationSpec destinationSpec;
+  private final Duration publishTimeout;
   private volatile ObservationCollector.ConnectionInfo connectionInfo;
 
   AmqpPublisher(AmqpPublisherBuilder builder) {
@@ -58,7 +61,9 @@ final class AmqpPublisher extends ResourceBase implements Publisher {
     this.address = builder.address();
     this.destinationSpec = builder.destination();
     this.connection = builder.connection();
-    this.sender = this.createSender(builder.connection().nativeSession(), this.address);
+    this.publishTimeout = builder.publishTimeout();
+    this.sender = this.createSender(builder.connection().nativeSession(), this.address,
+        this.publishTimeout);
     this.metricsCollector = this.connection.metricsCollector();
     this.observationCollector = this.connection.observationCollector();
     this.state(OPEN);
@@ -122,7 +127,8 @@ public void publish(Message message, Callback callback) {
 
   void recoverAfterConnectionFailure() {
     this.connectionInfo = new Utils.ObservationConnectionInfo(this.connection.connectionAddress());
-    this.sender = this.createSender(this.connection.nativeSession(false), this.address);
+    this.sender = this.createSender(this.connection.nativeSession(false), this.address,
+        this.publishTimeout);
   }
 
   @Override
@@ -132,8 +138,10 @@ public void close() {
 
   // internal API
 
-  private Sender createSender(Session session, String address) {
-    SenderOptions senderOptions = new SenderOptions().deliveryMode(DeliveryMode.AT_LEAST_ONCE);
+  private Sender createSender(Session session, String address, Duration publishTimeout) {
+    SenderOptions senderOptions = new SenderOptions().deliveryMode(DeliveryMode.AT_LEAST_ONCE)
+        .sendTimeout(publishTimeout.isNegative() ? ConnectionOptions.INFINITE :
+            publishTimeout.toMillis());
     try {
       if (address == null) {
         return session.openAnonymousSender(senderOptions);

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

@@ -21,6 +21,8 @@
 import com.rabbitmq.model.Publisher;
 import com.rabbitmq.model.PublisherBuilder;
 import com.rabbitmq.model.Resource;
+
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -30,6 +32,7 @@ class AmqpPublisherBuilder implements PublisherBuilder {
   private final List<Resource.StateListener> listeners = new ArrayList<>();
   private final DefaultAddressBuilder<PublisherBuilder> addressBuilder =
       new DefaultAddressBuilder<>(this) {};
+  private Duration publishTimeout = Duration.ofSeconds(60);
 
   AmqpPublisherBuilder(AmqpConnection connection) {
     this.connection = connection;
@@ -60,6 +63,12 @@ public PublisherBuilder listeners(Resource.StateListener... listeners) {
     return this;
   }
 
+
+  public AmqpPublisherBuilder publishTimeout(Duration publishTimeout) {
+    this.publishTimeout = publishTimeout;
+    return this;
+  }
+
   @Override
   public Publisher build() {
     return this.connection.createPublisher(this);
@@ -84,4 +93,8 @@ String address() {
   DefaultAddressBuilder.DestinationSpec destination() {
     return this.addressBuilder.destination();
   }
+
+  Duration publishTimeout() {
+    return this.publishTimeout;
+  }
 }