Add durableSubscription as a binding property

- This will be updated as a binder consumer property for the key `durableSubscription`

This resolves #293

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -85,6 +85,8 @@ public class BindingProperties {
 
 	private Integer concurrency;
 
+	private Boolean durableSubscription;
+
 	// Partition properties
 	private String partitionIndex;
 
@@ -243,6 +245,14 @@ public class BindingProperties {
 		this.partitioned = partitioned;
 	}
 
+	public Boolean getDurableSubscription() {
+		return this.durableSubscription;
+	}
+
+	public void setDurableSubscription(Boolean durableSubscription) {
+		this.durableSubscription = durableSubscription;
+	}
+
 	public String toString() {
 		StringBuilder sb = new StringBuilder();
 		sb.append("destination=" + this.destination);
@@ -315,6 +325,10 @@ public class BindingProperties {
 			sb.append("concurrency=" + this.concurrency);
 			sb.append(COMMA);
 		}
+		if (this.durableSubscription != null) {
+			sb.append("durableSubscription=" + this.durableSubscription);
+			sb.append(COMMA);
+		}
 		sb.deleteCharAt(sb.lastIndexOf(COMMA));
 		return "BindingProperties{" + sb.toString() + "}";
 	}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015 the original author or authors.
+ * Copyright 2015-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -112,6 +112,10 @@ public class ChannelBindingServiceProperties {
 				channelConsumerProperties.setProperty(BinderPropertyKeys.CONCURRENCY,
 						Integer.toString(bindingProperties.getConcurrency()));
 			}
+			if (bindingProperties.getDurableSubscription() != null) {
+				channelConsumerProperties.setProperty(BinderPropertyKeys.DURABLE,
+						Boolean.toString(bindingProperties.getDurableSubscription()));
+			}
 			updateConsumerPartitionProperties(inputChannelName, channelConsumerProperties);
 		}
 		return channelConsumerProperties;