Merge pull request #153 from alexpogue/extrahttpheaders

Add option for adding arbitrary headers to HTTP operators
IBMStreams · Sep 10, 2015 · c9a514a · c9a514a
2 parents deb8114 + 0e2c4bf
commit c9a514a
Show file tree

Hide file tree

Showing 5 changed files with 51 additions and 3 deletions.
diff --git a/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/AbstractHTTPGetContent.java b/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/AbstractHTTPGetContent.java
@@ -6,6 +6,8 @@
 
 import java.io.IOException;
 import java.net.HttpURLConnection;
+import java.util.List;
+import java.util.Map;
 import java.util.logging.Logger;
 
 import org.apache.http.HttpEntity;
@@ -35,6 +37,7 @@ public abstract class AbstractHTTPGetContent<A> extends
             .getLogger("com.ibm.streamsx.inet.http");
 
     private String url;
+    private List<String> extraHeaders;
 
     private HttpClient client;
     protected URIBuilder builder;
@@ -53,6 +56,11 @@ public String getUrl() {
     public void setUrl(String url) {
         this.url = url;
     }
+
+    @Parameter(optional = true, description = "Extra headers to send with request, format is \\\"Header-Name: value\\\"")
+    public void setExtraHeaders(List<String> extraHeaders) {
+    	this.extraHeaders = extraHeaders;
+    }
 
     public Metric getnFailedRequests() {
         return nFailedRequests;
@@ -85,6 +93,11 @@ public synchronized void initialize(OperatorContext context)
                     .getIndex();
         else
             contentAttributeIndex = defaultContentAttributeIndex();
+
+        Map<String, String> extraHeaderMap = HTTPUtils.getHeaderMap(extraHeaders);
+        for(Map.Entry<String, String> header : extraHeaderMap.entrySet()) {
+            get.setHeader(header.getKey(), header.getValue());
+        }
     }
 
     protected abstract String acceptedContentTypes();

diff --git a/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/HTTPPostOper.java b/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/HTTPPostOper.java
@@ -73,6 +73,8 @@ public class HTTPPostOper extends AbstractOperator
 
 	private String headerContentType = MIME_FORM;
 
+	private List<String> extraHeaders = new ArrayList<String>();
+
 	@Parameter(optional= false, description="URL to connect to")
 	public void setUrl(String url) {
 		this.url = url;
@@ -109,6 +111,10 @@ public void setRetryDelay(double val) {
 	public void setHeaderContentType(String val) {
 		this.headerContentType = val;
 	}
+	@Parameter(optional=true, description="Extra headers to send with request, format is \\\"Header-Name: value\\\".")
+	public void setExtraHeaders(List<String> val) {
+		this.extraHeaders = val;
+	}
 
 	//consistent region checks
 	@ContextCheck(compile = true)
@@ -175,6 +181,11 @@ else if(headerContentType.equals(MIME_JSON)){
 			req.setParams(tuple.getObject(schema.getAttribute(0).getName()).toString());
 		}
 
+		Map<String, String> headerMap = HTTPUtils.getHeaderMap(extraHeaders);
+		for(Map.Entry<String, String> header : headerMap.entrySet()) {
+			req.setHeader(header.getKey(), header.getValue());
+		}
+
 		HTTPResponse resp = null;
 
 		Throwable t = null;

diff --git a/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/HTTPStreamReader.java b/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/HTTPStreamReader.java
@@ -58,6 +58,7 @@ public class HTTPStreamReader extends AbstractOperator {
 	private String authenticationType = "none", authenticationFile = null;
 	private RetryController rc = null;
 	private List<String> authenticationProperties = new ArrayList<String>();
+	private List<String> extraHeaders = new ArrayList<String>();
 
 	private static Logger trace = Logger.getLogger(CLASS_NAME);
 	private boolean retryOnClose = false;
@@ -118,6 +119,10 @@ public void setRetryOnClose(boolean val) {
 	public void setDisableCompression(boolean val) {
 		this.disableCompression = val;
 	}
+	@Parameter(optional=true, description="Extra headers to send with request, format is \\\"Header-Name: value\\\".")
+	public void setExtraHeaders(List<String> val) {
+		this.extraHeaders = val;
+	}
 
 	@ContextCheck(compile=true)
 	public static boolean checkAuthParams(OperatorContextChecker occ) {
@@ -180,10 +185,12 @@ public void initialize(OperatorContext op) throws Exception {
 		if(authenticationFile != null) {
             authenticationFile = authenticationFile.trim();
         }
+
         URI baseConfigURI = op.getPE().getApplicationDirectory().toURI();
 		IAuthenticate auth = AuthHelper.getAuthenticator(authenticationType, PathConversionHelper.convertToAbsPath(baseConfigURI, authenticationFile), authenticationProperties);
-
-		reader = new HTTPStreamReaderObj(this.url, auth, this, postDataParams, disableCompression);
+		Map<String, String> extraHeaderMap = HTTPUtils.getHeaderMap(extraHeaders);
+
+		reader = new HTTPStreamReaderObj(this.url, auth, this, postDataParams, disableCompression, extraHeaderMap);
 		th = op.getThreadFactory().newThread(reader);
 		th.setDaemon(false);
 	}

diff --git a/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/HTTPStreamReaderObj.java b/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/HTTPStreamReaderObj.java
@@ -48,7 +48,7 @@ class HTTPStreamReaderObj implements Runnable
 
 	public HTTPStreamReaderObj(String url, IAuthenticate auth, 
 			HTTPStreamReader reader,  Map<String, String> postD, 
-			boolean disableCompression) 
+			boolean disableCompression, Map<String, String> extraHeaders) 
 			throws Exception 	{
 		this.auth = auth;
 		this.reader = reader;
@@ -63,6 +63,9 @@ public HTTPStreamReaderObj(String url, IAuthenticate auth,
 		else {
 			req.setHeader("Accept-Encoding", "identity");
 		}
+		for(Map.Entry<String, String> header : extraHeaders.entrySet()) {
+			req.setHeader(header.getKey(), header.getValue());
+		}
 		req.setParams(postData);
 	}
 

diff --git a/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/HTTPUtils.java b/com.ibm.streamsx.inet/impl/java/src/com/ibm/streamsx/inet/http/HTTPUtils.java
@@ -12,6 +12,9 @@
 import java.io.InputStreamReader;
 import java.net.HttpURLConnection;
 import java.net.URL;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 public class HTTPUtils {
 
@@ -39,4 +42,15 @@ public static String readFromStream(InputStream stream) {
 		}
 		return buf.toString();
 	}
+
+	public static Map<String, String> getHeaderMap(List<String> headers) {
+		Map<String, String> headerMap = new HashMap<String, String>(headers.size());
+		for(String header : headers) {
+			String[] headerParts = header.split(":\\s*", 2);
+			String headerName = headerParts[0];
+			String headerValue = headerParts[1];
+			headerMap.put(headerName, headerValue);
+		}
+		return headerMap;
+	}
 }