You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by dl...@apache.org on 2016/06/03 18:02:56 UTC
[04/11] incubator-quarks git commit: implemented review comments
implemented review comments
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/3c4cd8fc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/3c4cd8fc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/3c4cd8fc
Branch: refs/heads/master
Commit: 3c4cd8fc008f93b367620e9958de50c1686a10a2
Parents: c9d9342
Author: Saurabh Jinturkar <sa...@sjsu.edu>
Authored: Wed May 11 01:09:44 2016 -0700
Committer: Saurabh Jinturkar <sa...@sjsu.edu>
Committed: Wed May 11 01:09:44 2016 -0700
----------------------------------------------------------------------
.../quarks/connectors/http/HttpStreams.java | 30 +++-
.../connectors/http/runtime/HttpRequester.java | 1 +
.../quarks/test/connectors/http/HttpTest.java | 164 ++++++++++++++++++-
3 files changed, 189 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/3c4cd8fc/connectors/http/src/main/java/quarks/connectors/http/HttpStreams.java
----------------------------------------------------------------------
diff --git a/connectors/http/src/main/java/quarks/connectors/http/HttpStreams.java b/connectors/http/src/main/java/quarks/connectors/http/HttpStreams.java
index 2b7e088..a22c760 100644
--- a/connectors/http/src/main/java/quarks/connectors/http/HttpStreams.java
+++ b/connectors/http/src/main/java/quarks/connectors/http/HttpStreams.java
@@ -21,8 +21,11 @@ package quarks.connectors.http;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpDelete;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import com.google.gson.JsonObject;
@@ -31,6 +34,7 @@ import quarks.connectors.http.runtime.HttpRequester;
import quarks.function.BiFunction;
import quarks.function.Function;
import quarks.function.Supplier;
+import quarks.function.UnaryOperator;
import quarks.topology.TStream;
@@ -52,6 +56,7 @@ public class HttpStreams {
*
* <pre>
* {@code
+ * final String url = "http://httpbin.org/get?";
* JsonObject request1 = new JsonObject();
* request1.addProperty("a", "abc");
* request1.addProperty("b", "42");
@@ -80,6 +85,14 @@ public class HttpStreams {
t -> HttpGet.METHOD_NAME, uri, HttpResponders.json());
}
+ public static TStream<JsonObject> deleteJson(TStream<JsonObject> stream,
+ Supplier<CloseableHttpClient> clientCreator,
+ Function<JsonObject,String> uri) {
+
+ return HttpStreams.<JsonObject,JsonObject>requests(stream, clientCreator,
+ t -> HttpDelete.METHOD_NAME, uri, HttpResponders.json());
+ }
+
/**
* Make an HTTP POST request with JsonObject. <br>
*
@@ -92,6 +105,7 @@ public class HttpStreams {
*
* <pre>
* {@code
+ * final String url = "http://httpbin.org/post";
* JsonObject request = new JsonObject();
* request.addProperty("a", "abc");
* request.addProperty("b", "42");
@@ -117,10 +131,22 @@ public class HttpStreams {
public static TStream<JsonObject> postJson(TStream<JsonObject> stream,
Supplier<CloseableHttpClient> clientCreator,
Function<JsonObject, String> uri,
- Function<JsonObject, HttpEntity> body) {
+ UnaryOperator<JsonObject> body) {
+
+ return HttpStreams.<JsonObject, JsonObject> requestsWithBody(stream,
+ clientCreator, t -> HttpPost.METHOD_NAME, uri,
+ t -> new ByteArrayEntity(body.apply(t).toString().getBytes()),
+ HttpResponders.json());
+ }
+
+ public static TStream<JsonObject> putJson(TStream<JsonObject> stream,
+ Supplier<CloseableHttpClient> clientCreator,
+ Function<JsonObject, String> uri,
+ UnaryOperator<JsonObject> body) {
return HttpStreams.<JsonObject, JsonObject> requestsWithBody(stream,
- clientCreator, t -> HttpPost.METHOD_NAME, uri, body,
+ clientCreator, t -> HttpPut.METHOD_NAME, uri,
+ t -> new ByteArrayEntity(body.apply(t).toString().getBytes()),
HttpResponders.json());
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/3c4cd8fc/connectors/http/src/main/java/quarks/connectors/http/runtime/HttpRequester.java
----------------------------------------------------------------------
diff --git a/connectors/http/src/main/java/quarks/connectors/http/runtime/HttpRequester.java b/connectors/http/src/main/java/quarks/connectors/http/runtime/HttpRequester.java
index 36b5411..88b560a 100644
--- a/connectors/http/src/main/java/quarks/connectors/http/runtime/HttpRequester.java
+++ b/connectors/http/src/main/java/quarks/connectors/http/runtime/HttpRequester.java
@@ -99,6 +99,7 @@ public class HttpRequester<T,R> implements Function<T,R>{
case HttpPut.METHOD_NAME:
request = new HttpPut(uri);
break;
+
default:
throw new IllegalArgumentException();
}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/3c4cd8fc/connectors/http/src/test/java/quarks/test/connectors/http/HttpTest.java
----------------------------------------------------------------------
diff --git a/connectors/http/src/test/java/quarks/test/connectors/http/HttpTest.java b/connectors/http/src/test/java/quarks/test/connectors/http/HttpTest.java
index ddad71e..c30e3ea 100644
--- a/connectors/http/src/test/java/quarks/test/connectors/http/HttpTest.java
+++ b/connectors/http/src/test/java/quarks/test/connectors/http/HttpTest.java
@@ -25,7 +25,10 @@ import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
+import org.apache.http.client.methods.HttpDelete;
import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.ByteArrayEntity;
import org.junit.Test;
@@ -73,6 +76,77 @@ public class HttpTest {
}
@Test
+ public void testPost() throws Exception {
+ DirectProvider ep = new DirectProvider();
+
+ Topology topology = ep.newTopology();
+
+ String url = "http://httpbin.org/post";
+
+ TStream<String> rc = HttpStreams.<String, String>requestsWithBody(
+ topology.strings(url),
+ HttpClients::noAuthentication,
+ t -> HttpPost.METHOD_NAME,
+ t-> t, t-> new ByteArrayEntity(t.getBytes()),
+ HttpResponders.inputOn200());
+
+ Tester tester = topology.getTester();
+
+ Condition<List<String>> endCondition = tester.streamContents(rc, url);
+
+ tester.complete(ep, new JsonObject(), endCondition, 10, TimeUnit.SECONDS);
+
+ assertTrue(endCondition.valid());
+ }
+
+ @Test
+ public void testPut() throws Exception {
+ DirectProvider ep = new DirectProvider();
+
+ Topology topology = ep.newTopology();
+
+ String url = "http://httpbin.org/put";
+
+ TStream<String> rc = HttpStreams.<String, String>requestsWithBody(
+ topology.strings(url),
+ HttpClients::noAuthentication,
+ t -> HttpPut.METHOD_NAME,
+ t-> t, t-> new ByteArrayEntity(t.getBytes()),
+ HttpResponders.inputOn200());
+
+ Tester tester = topology.getTester();
+
+ Condition<List<String>> endCondition = tester.streamContents(rc, url);
+
+ tester.complete(ep, new JsonObject(), endCondition, 10, TimeUnit.SECONDS);
+
+ assertTrue(endCondition.valid());
+ }
+
+ @Test
+ public void testDelete() throws Exception {
+ DirectProvider ep = new DirectProvider();
+
+ Topology topology = ep.newTopology();
+
+ String url = "http://httpbin.org/delete";
+
+ TStream<String> rc = HttpStreams.<String, String>requests(
+ topology.strings(url),
+ HttpClients::noAuthentication,
+ t -> HttpDelete.METHOD_NAME,
+ t-> t, HttpResponders.inputOn200());
+
+ Tester tester = topology.getTester();
+
+ Condition<List<String>> endCondition = tester.streamContents(rc, url);
+
+ tester.complete(ep, new JsonObject(), endCondition, 10, TimeUnit.SECONDS);
+
+ assertTrue(endCondition.valid());
+ }
+
+ @Test
public void testStatusCode() throws Exception {
DirectProvider ep = new DirectProvider();
@@ -173,22 +247,104 @@ public class HttpTest {
}
@Test
+ public void testJsonDelete() throws Exception {
+
+ DirectProvider ep = new DirectProvider();
+
+ Topology topology = ep.newTopology();
+
+ final String url = "http://httpbin.org/delete?";
+
+ JsonObject request1 = new JsonObject();
+ request1.addProperty("a", "abc");
+ request1.addProperty("b", "42");
+
+ TStream<JsonObject> rc = HttpStreams.deleteJson(
+ topology.collection(Arrays.asList(request1)),
+ HttpClients::noAuthentication,
+ t-> url + "a=" + t.get("a").getAsString() + "&b=" + t.get("b").getAsString()
+ );
+
+ TStream<Boolean> resStream = rc.map(j -> {
+ assertTrue(j.has("request"));
+ assertTrue(j.has("response"));
+ JsonObject req = j.getAsJsonObject("request");
+ JsonObject res = j.getAsJsonObject("response");
+
+ assertTrue(res.has("status"));
+ assertTrue(res.has("entity"));
+
+ assertEquals(req, res.getAsJsonObject("entity").getAsJsonObject("args"));
+ return true;
+ }
+ );
+
+ rc.print();
+
+ Tester tester = topology.getTester();
+
+ Condition<List<Boolean>> endCondition = tester.streamContents(resStream, true);
+
+ tester.complete(ep, new JsonObject(), endCondition, 10, TimeUnit.SECONDS);
+
+ assertTrue(endCondition.getResult().toString(), endCondition.valid());
+ }
+
+ @Test
public void testJsonPost() throws Exception {
DirectProvider ep = new DirectProvider();
Topology topology = ep.newTopology();
- final String url = "http://httpbin.org/post?";
+ final String url = "http://httpbin.org/post";
JsonObject request = new JsonObject();
- request.addProperty("a", "abc");
- request.addProperty("b", 42);
+ request.addProperty("foo", "abc");
+ request.addProperty("bar", 42);
TStream<JsonObject> rc = HttpStreams.postJson(
topology.collection(Arrays.asList(request)),
HttpClients::noAuthentication, t -> url,
- t -> new ByteArrayEntity(request.toString().getBytes()));
+ t -> request);
+
+ TStream<Boolean> resStream = rc.map(j -> {
+ assertTrue(j.has("request"));
+ assertTrue(j.has("response"));
+ JsonObject req = j.getAsJsonObject("request");
+ JsonObject res = j.getAsJsonObject("response");
+
+ assertTrue(res.has("status"));
+ assertTrue(res.has("entity"));
+
+ assertEquals(req, res.getAsJsonObject("entity").getAsJsonObject("json"));
+ return true;
+ });
+
+ rc.print();
+ Tester tester = topology.getTester();
+ Condition<List<Boolean>> endCondition = tester.streamContents(resStream, true);
+ tester.complete(ep, new JsonObject(), endCondition, 10, TimeUnit.SECONDS);
+ assertTrue(endCondition.getResult().toString(), endCondition.valid());
+ }
+
+ @Test
+ public void testJsonPut() throws Exception {
+
+ DirectProvider ep = new DirectProvider();
+
+ Topology topology = ep.newTopology();
+
+ final String url = "http://httpbin.org/put";
+
+ JsonObject request = new JsonObject();
+ request.addProperty("foo", "abc");
+ request.addProperty("bar", 42);
+
+ TStream<JsonObject> rc = HttpStreams.putJson(
+ topology.collection(Arrays.asList(request)),
+ HttpClients::noAuthentication, t -> url,
+ t -> request);
TStream<Boolean> resStream = rc.map(j -> {
assertTrue(j.has("request"));