HTTPTupleView Windowing does not support multiple partition keys in a…

… comma-delimited string #324

com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/rest/ops/TupleView.java

@@ -25,7 +25,8 @@
 public class TupleView extends ServletOperator {
     static final String opName = "HTTPTupleView";
 	// Parameter setters just to define the parameters in the SPL operator model.
-	@Parameter(optional=true, cardinality=-1, description="Names of attributes to partition the window by")
+	@Parameter(optional=true, cardinality=-1, description="Names of attributes to partition the window by. If the cardinality of this parameter is > 1,"
+			+ "then every value represents one attribute name. If the cadinality equals to 1, the value may contain one attribute name or a comma separated list of attribute names.")
 	public void setPartitionKey(String[] attributeNames) {}
 
     @Parameter(optional = true, description = "List of headers to insert into the http reply. Formatted as header:value")

com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/window/WindowContentsAtTrigger.java

@@ -42,11 +42,26 @@ public class WindowContentsAtTrigger<T> implements StreamWindowListener<T> {
 	private long lastModified = System.currentTimeMillis();
 
 	@SuppressWarnings("unchecked")
-    public WindowContentsAtTrigger(OperatorContext context, StreamingInput<T> input) {
+	public WindowContentsAtTrigger(OperatorContext context, StreamingInput<T> input) {
 		this.context = context;
 		this.input = input;
 		isSliding = StreamWindow.Type.SLIDING.equals(input.getStreamWindow().getType());
-		List<String> partitionKeys = context.getParameterValues("partitionKey");
+		List<String> partitionKeys = null;
+		//Tokenize paritionKey if cardinality is 1 and put all values to partitionKeys
+		List<String> primaryPartitionKeys = context.getParameterValues("partitionKey");
+		if (primaryPartitionKeys.size() == 1) {
+			String[] stringArr = primaryPartitionKeys.get(0).split(",");
+			if (stringArr.length > 1) {
+				partitionKeys = new ArrayList<String>();
+				for (int i = 0; i < stringArr.length; i++) {
+					partitionKeys.add(stringArr[i]);
+				}
+			} else {
+				partitionKeys = primaryPartitionKeys;
+			}
+		} else {
+			partitionKeys = primaryPartitionKeys;
+		}
 		if (!partitionKeys.isEmpty()) {
 		    if (!input.getStreamWindow().isPartitioned())
 		        throw new IllegalStateException("Input port " + input.getName() + "is not partitioned");