From 95c2f7ee4991f505ffefdd15d59c27451aea209b Mon Sep 17 00:00:00 2001
From: Gary Russell <grussell@pivotal.io>
Date: Mon, 11 Jun 2018 17:39:17 -0400
Subject: [PATCH] GH-155: Support overflowBehavior

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/155

Add `overFlowBehavior` and `dlqOverflowBehavior` queue properties.

Resolves #158
---
 .../properties/RabbitCommonProperties.java    | 26 +++++++++++++++++++
 .../RabbitExchangeQueueProvisioner.java       | 19 +++++++++-----
 .../src/main/asciidoc/overview.adoc           |  8 ++++++
 .../binder/rabbit/RabbitBinderTests.java      |  4 +++
 4 files changed, 50 insertions(+), 7 deletions(-)

diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java
index 1dcc5c425..a30709a5f 100644
--- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java
+++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java
@@ -164,6 +164,16 @@ public abstract class RabbitCommonProperties {
 	 */
 	private boolean dlqLazy;
 
+	/**
+	 * action when maxLength or maxLengthBytes is exceeded
+	 */
+	private String overflowBehavior;
+
+	/**
+	 * action when maxLength or maxLengthBytes is exceeded
+	 */
+	private String dlqOverflowBehavior;
+
 	public String getExchangeType() {
 		return this.exchangeType;
 	}
@@ -381,4 +391,20 @@ public abstract class RabbitCommonProperties {
 		this.dlqLazy = dlqLazy;
 	}
 
+	public String getOverflowBehavior() {
+		return this.overflowBehavior;
+	}
+
+	public void setOverflowBehavior(String overflowBehavior) {
+		this.overflowBehavior = overflowBehavior;
+	}
+
+	public String getDlqOverflowBehavior() {
+		return this.dlqOverflowBehavior;
+	}
+
+	public void setDlqOverflowBehavior(String dlqOverflowBehavior) {
+		this.dlqOverflowBehavior = dlqOverflowBehavior;
+	}
+
 }
diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
index 67706d5fb..9c70f8894 100644
--- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
+++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
@@ -380,8 +380,6 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 				}
 				args.put("x-dead-letter-routing-key", dlRk);
 			}
-			additionalArgs(args, properties.getExpires(), properties.getMaxLength(), properties.getMaxLengthBytes(),
-					properties.getMaxPriority(), properties.getTtl(), properties.isLazy());
 		}
 		else {
 			if (properties.getDlqDeadLetterExchange() != null) {
@@ -390,15 +388,19 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 			if (properties.getDlqDeadLetterRoutingKey() != null) {
 				args.put("x-dead-letter-routing-key", properties.getDlqDeadLetterRoutingKey());
 			}
-			additionalArgs(args, properties.getDlqExpires(), properties.getDlqMaxLength(),
-					properties.getDlqMaxLengthBytes(), properties.getDlqMaxPriority(), properties.getDlqTtl(),
-					properties.isDlqLazy());
 		}
+		additionalArgs(args, properties, isDlq);
 		return args;
 	}
 
-	private void additionalArgs(Map<String, Object> args, Integer expires, Integer maxLength, Integer maxLengthBytes,
-								Integer maxPriority, Integer ttl, boolean lazy) {
+	private void additionalArgs(Map<String, Object> args, RabbitCommonProperties properties, boolean isDlq) {
+		Integer expires = isDlq ? properties.getDlqExpires() : properties.getExpires();
+		Integer maxLength = isDlq ? properties.getDlqMaxLength() : properties.getMaxLength();
+		Integer maxLengthBytes = isDlq ? properties.getDlqMaxLengthBytes() : properties.getMaxLengthBytes();
+		Integer maxPriority = isDlq ? properties.getDlqMaxPriority() : properties.getMaxPriority();
+		Integer ttl = isDlq ? properties.getDlqTtl() : properties.getTtl();
+		boolean lazy = isDlq ? properties.isDlqLazy() : properties.isLazy();
+		String overflow = isDlq ? properties.getDlqOverflowBehavior() : properties.getOverflowBehavior();
 		if (expires != null) {
 			args.put("x-expires", expires);
 		}
@@ -417,6 +419,9 @@ public class RabbitExchangeQueueProvisioner implements ApplicationListener<Decla
 		if (lazy) {
 			args.put("x-queue-mode", "lazy");
 		}
+		if (StringUtils.hasText(overflow)) {
+			args.put("x-overflow", overflow);
+		}
 	}
 
 
diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc
index 4ffa57b44..0de58b736 100644
--- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc
+++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc
@@ -184,6 +184,10 @@ dlqMaxPriority::
 Maximum priority of messages in the dead letter queue (0-255).
 +
 Default: `none`
+dlqOverflowBehavior::
+Action to take when `dlqMaxLength` or `dlqMaxLengthBytes` is exceeded; currently `drop-head` or `reject-publish` but refer to the RabbitMQ documentation.
++
+Default: `none`
 dlqTtl::
 Default time to live to apply to the dead letter queue when declared (in milliseconds).
 +
@@ -251,6 +255,10 @@ When the queue cannot be found, whether to treat the condition as fatal and stop
 Defaults to `false` so that the container keeps trying to consume from the queue -- for example, when using a cluster and the node hosting a non-HA queue is down.
 +
 Default: `false`
+overflowBehavior::
+Action to take when `maxLength` or `maxLengthBytes` is exceeded; currently `drop-head` or `reject-publish` but refer to the RabbitMQ documentation.
++
+Default: `none`
 prefetch::
 Prefetch count.
 +
diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
index 0dbc62fb1..e5de8a3ab 100644
--- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
+++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
@@ -416,6 +416,7 @@ public class RabbitBinderTests extends
 		extProps.setMaxLength(10_000);
 		extProps.setMaxLengthBytes(100_000);
 		extProps.setMaxPriority(10);
+		extProps.setOverflowBehavior("drop-head");
 		extProps.setTtl(2_000);
 		extProps.setAutoBindDlq(true);
 		extProps.setDeadLetterQueueName("customDLQ");
@@ -427,6 +428,7 @@ public class RabbitBinderTests extends
 		extProps.setDlqLazy(true);
 		extProps.setDlqMaxLength(20_000);
 		extProps.setDlqMaxLengthBytes(40_000);
+		extProps.setDlqOverflowBehavior("reject-publish");
 		extProps.setDlqMaxPriority(8);
 		extProps.setDlqTtl(1_000);
 
@@ -472,6 +474,7 @@ public class RabbitBinderTests extends
 		assertThat(args.get("x-expires")).isEqualTo(30_000);
 		assertThat(args.get("x-max-length")).isEqualTo(10_000);
 		assertThat(args.get("x-max-length-bytes")).isEqualTo(100_000);
+		assertThat(args.get("x-overflow")).isEqualTo("drop-head");
 		assertThat(args.get("x-max-priority")).isEqualTo(10);
 		assertThat(args.get("x-message-ttl")).isEqualTo(2_000);
 		assertThat(args.get("x-dead-letter-exchange")).isEqualTo("customDLX");
@@ -490,6 +493,7 @@ public class RabbitBinderTests extends
 		assertThat(args.get("x-expires")).isEqualTo(60_000);
 		assertThat(args.get("x-max-length")).isEqualTo(20_000);
 		assertThat(args.get("x-max-length-bytes")).isEqualTo(40_000);
+		assertThat(args.get("x-overflow")).isEqualTo("reject-publish");
 		assertThat(args.get("x-max-priority")).isEqualTo(8);
 		assertThat(args.get("x-message-ttl")).isEqualTo(1_000);
 		assertThat(args.get("x-dead-letter-exchange")).isEqualTo("propsUser3");