Merge pull request #1 from 01org/test-new-api
Upgrade to new API and add SegmentedServerHelper
diff --git a/pom.xml b/pom.xml
index ecd5e24..c3144df 100644
--- a/pom.xml
+++ b/pom.xml
@@ -3,7 +3,7 @@
<modelVersion>4.0.0</modelVersion>
<groupId>com.intel.jndn.utils</groupId>
<artifactId>jndn-utils</artifactId>
- <version>0.9.6</version>
+ <version>0.9.7</version>
<name>jndn-utils</name>
<description>Collection of tools to simplify synchronous and asynchronous data transfer over the NDN network</description>
<url>https://github.com/01org/jndn-utils</url>
diff --git a/src/main/java/com/intel/jndn/utils/Client.java b/src/main/java/com/intel/jndn/utils/Client.java
index cb90f6f..6b09f9f 100644
--- a/src/main/java/com/intel/jndn/utils/Client.java
+++ b/src/main/java/com/intel/jndn/utils/Client.java
@@ -14,86 +14,28 @@
package com.intel.jndn.utils;
import java.io.IOException;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeoutException;
-import java.util.logging.Level;
+import java.util.concurrent.Future;
import net.named_data.jndn.Data;
import net.named_data.jndn.Face;
import net.named_data.jndn.Interest;
-import net.named_data.jndn.Name;
-import net.named_data.jndn.OnData;
-import net.named_data.jndn.OnTimeout;
-import java.util.logging.Logger;
/**
- * Provide a client to simplify information retrieval over the NDN network.
+ * Base functionality provided by all NDN clients in this package.
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
-public class Client {
-
- public static final long DEFAULT_SLEEP_TIME = 20;
- public static final long DEFAULT_TIMEOUT = 2000;
- private static final Logger logger = Logger.getLogger(Client.class.getName());
- private static Client defaultInstance;
-
- /**
- * Singleton access for simpler client use
- *
- * @return
- */
- public static Client getDefault() {
- if (defaultInstance == null) {
- defaultInstance = new Client();
- }
- return defaultInstance;
- }
+public interface Client {
/**
* Asynchronously request the Data for an Interest. This will send the
* Interest and return immediately; use futureData.get() to block until the
- * Data returns (see FutureData) or manage the event processing independently.
+ * Data returns (see Future) or manage the event processing independently.
*
* @param face
* @param interest
* @return
*/
- public FutureData getAsync(Face face, Interest interest) {
- final FutureData futureData = new FutureData(face, interest.getName());
-
- // send interest
- try {
- face.expressInterest(interest, new OnData() {
- @Override
- public void onData(Interest interest, Data data) {
- futureData.resolve(data);
- }
- }, new OnTimeout() {
- @Override
- public void onTimeout(Interest interest) {
- futureData.reject(new TimeoutException());
- }
- });
- } catch (IOException e) {
- logger.log(Level.WARNING, "IO failure while sending interest: ", e);
- futureData.reject(e);
- }
-
- return futureData;
- }
-
- /**
- * Synchronously retrieve the Data for a Name using a default interest (e.g. 2
- * second timeout); this will block until complete (i.e. either data is
- * received or the interest times out).
- *
- * @param face
- * @param name
- * @return
- */
- public FutureData getAsync(Face face, Name name) {
- return getAsync(face, getDefaultInterest(name));
- }
+ public Future<Data> getAsync(Face face, Interest interest);
/**
* Synchronously retrieve the Data for an Interest; this will block until
@@ -101,48 +43,8 @@
*
* @param face
* @param interest
- * @return Data packet or null
- */
- public Data getSync(Face face, Interest interest) {
- // setup event
- long startTime = System.currentTimeMillis();
-
- // get future data
- FutureData futureData = getAsync(face, interest);
-
- // process eventCount until a response is received or timeout
- try {
- Data data = futureData.get();
- logger.fine("Request time (ms): " + (System.currentTimeMillis() - startTime));
- return data;
- } catch (ExecutionException | InterruptedException e) {
- logger.log(Level.WARNING, "Failed to retrieve data: ", e);
- return null;
- }
- }
-
- /**
- * Synchronously retrieve the Data for a Name using a default interest (e.g. 2
- * second timeout); this will block until complete (i.e. either data is
- * received or the interest times out).
- *
- * @param face
- * @param name
* @return
+ * @throws java.io.IOException
*/
- public Data getSync(Face face, Name name) {
- return getSync(face, getDefaultInterest(name));
- }
-
- /**
- * Create a default interest for a given Name using some common settings: -
- * lifetime: 2 seconds
- *
- * @param name
- * @return
- */
- public static Interest getDefaultInterest(Name name) {
- Interest interest = new Interest(name, DEFAULT_TIMEOUT);
- return interest;
- }
+ public Data getSync(Face face, Interest interest) throws IOException;
}
diff --git a/src/main/java/com/intel/jndn/utils/SegmentedClient.java b/src/main/java/com/intel/jndn/utils/SegmentedClient.java
index 38524ef..40ef7bb 100644
--- a/src/main/java/com/intel/jndn/utils/SegmentedClient.java
+++ b/src/main/java/com/intel/jndn/utils/SegmentedClient.java
@@ -13,11 +13,13 @@
*/
package com.intel.jndn.utils;
-import java.io.ByteArrayOutputStream;
+import com.intel.jndn.utils.client.SegmentedFutureData;
+import com.intel.jndn.utils.client.FutureData;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
import java.util.logging.Level;
import java.util.logging.Logger;
import net.named_data.jndn.Data;
@@ -25,7 +27,6 @@
import net.named_data.jndn.Interest;
import net.named_data.jndn.Name;
import net.named_data.jndn.encoding.EncodingException;
-import net.named_data.jndn.util.Blob;
/**
* Provide a client to simplify retrieving segmented Data packets over the NDN
@@ -38,14 +39,13 @@
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
-public class SegmentedClient {
+public class SegmentedClient implements Client {
private static SegmentedClient defaultInstance;
private static final Logger logger = Logger.getLogger(SegmentedClient.class.getName());
- private static final int SLEEP_TIME_MS = 10;
/**
- * Singleton access for simpler client use
+ * Singleton access for simpler client use.
*
* @return
*/
@@ -67,7 +67,37 @@
* @return a list of FutureData packets; if the first segment fails, the list
* will contain one FutureData with the failure exception
*/
- public List<FutureData> getAsync(Face face, Interest interest) {
+ @Override
+ public Future<Data> getAsync(Face face, Interest interest) {
+ List<Future<Data>> segments = getAsyncList(face, interest);
+ return new SegmentedFutureData(interest.getName().getPrefix(-1), segments);
+ }
+
+ /**
+ * Asynchronously send Interest packets for a segmented result; will block
+ * until the first packet is received and then send remaining interests until
+ * the specified FinalBlockId.
+ *
+ * @param face
+ * @param name
+ * @return an aggregated data packet from all received segments
+ */
+ public Future<Data> getAsync(Face face, Name name) {
+ return getAsync(face, SimpleClient.getDefaultInterest(name));
+ }
+
+ /**
+ * Asynchronously send Interest packets for a segmented result; will block
+ * until the first packet is received and then send remaining interests until
+ * the specified FinalBlockId.
+ *
+ * @param face
+ * @param interest should include either a ChildSelector or an initial segment
+ * number
+ * @return a list of FutureData packets; if the first segment fails, the list
+ * will contain one FutureData with the failure exception
+ */
+ public List<Future<Data>> getAsyncList(Face face, Interest interest) {
// get first segment; default 0 or use a specified start segment
long firstSegment = 0;
boolean specifiedSegment = false;
@@ -78,12 +108,13 @@
// check for interest selector if no initial segment found
if (interest.getChildSelector() == -1) {
logger.log(Level.WARNING, "No child selector set for a segmented Interest; this may result in incorrect retrieval.");
+ // allow this interest to pass without a segment marker since it may still succeed
}
}
// setup segments
- final List<FutureData> segments = new ArrayList<>();
- segments.add(Client.getDefault().getAsync(face, interest));
+ final List<Future<Data>> segments = new ArrayList<>();
+ segments.add(SimpleClient.getDefault().getAsync(face, interest));
// retrieve first packet to find last segment value
long lastSegment;
@@ -103,7 +134,7 @@
for (long i = firstSegment + 1; i <= lastSegment; i++) {
Interest segmentedInterest = new Interest(interest);
segmentedInterest.getName().appendSegment(i);
- FutureData futureData = Client.getDefault().getAsync(face, segmentedInterest);
+ Future<Data> futureData = SimpleClient.getDefault().getAsync(face, segmentedInterest);
segments.add((int) i, futureData);
}
@@ -120,8 +151,8 @@
* @param name
* @return
*/
- public List<FutureData> getAsync(Face face, Name name) {
- return getAsync(face, Client.getDefaultInterest(name));
+ public List<Future<Data>> getAsyncList(Face face, Name name) {
+ return getAsyncList(face, SimpleClient.getDefaultInterest(name));
}
/**
@@ -132,40 +163,18 @@
* @param interest should include either a ChildSelector or an initial segment
* number
* @return a Data packet; the name will inherit from the sent Interest, not
- * the returned packets (TODO or should we parse from returned Data? copy
- * first Data?) and the content will be a concatenation of all of the packet
- * contents.
+ * the returned packets and the content will be a concatenation of all of the
+ * packet contents.
+ * @throws java.io.IOException
*/
- public Data getSync(Face face, Interest interest) {
- List<FutureData> segments = getAsync(face, interest);
-
- // process events until complete
- while (!isSegmentListComplete(segments)) {
- try {
- face.processEvents();
- Thread.sleep(SLEEP_TIME_MS);
- } catch (EncodingException | IOException e) {
- logger.log(Level.WARNING, "Failed to retrieve data: ", e);
- return null;
- } catch (InterruptedException ex) {
- // do nothing
- }
+ @Override
+ public Data getSync(Face face, Interest interest) throws IOException {
+ try {
+ return getAsync(face, interest).get();
+ } catch (ExecutionException | InterruptedException e) {
+ logger.log(Level.WARNING, "Failed to retrieve data.", e);
+ throw new IOException("Failed to retrieve data.", e);
}
-
- // build final blob
- ByteArrayOutputStream content = new ByteArrayOutputStream();
- for (FutureData futureData : segments) {
- try {
- content.write(futureData.get().getContent().getImmutableArray());
- } catch (ExecutionException | IOException | InterruptedException e) {
- logger.log(Level.WARNING, "Failed to parse retrieved data: ", e);
- return null;
- }
- }
-
- Data data = new Data(interest.getName()); // TODO this name may not be correct; may need to contain additional suffixes
- data.setContent(new Blob(content.toByteArray()));
- return data;
}
/**
@@ -177,9 +186,10 @@
* @param face
* @param name
* @return
+ * @throws java.io.IOException
*/
- public Data getSync(Face face, Name name) {
- return getSync(face, Client.getDefaultInterest(name));
+ public Data getSync(Face face, Name name) throws IOException {
+ return getSync(face, SimpleClient.getDefaultInterest(name));
}
/**
@@ -193,19 +203,4 @@
public static boolean hasSegment(Name name) {
return name.get(-1).getValue().buf().get(0) == 0x00;
}
-
- /**
- * Check if a list of segments have returned from the network.
- *
- * @param segments
- * @return
- */
- protected boolean isSegmentListComplete(List<FutureData> segments) {
- for (FutureData futureData : segments) {
- if (!futureData.isDone()) {
- return false;
- }
- }
- return true;
- }
}
diff --git a/src/main/java/com/intel/jndn/utils/SimpleClient.java b/src/main/java/com/intel/jndn/utils/SimpleClient.java
new file mode 100644
index 0000000..a751f8e
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/SimpleClient.java
@@ -0,0 +1,145 @@
+/*
+ * jndn-utils
+ * Copyright (c) 2015, Intel Corporation.
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms and conditions of the GNU Lesser General Public License,
+ * version 3, as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope it will be useful, but WITHOUT ANY
+ * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
+ * more details.
+ */
+package com.intel.jndn.utils;
+
+import com.intel.jndn.utils.client.FutureData;
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeoutException;
+import java.util.logging.Level;
+import net.named_data.jndn.Data;
+import net.named_data.jndn.Face;
+import net.named_data.jndn.Interest;
+import net.named_data.jndn.Name;
+import net.named_data.jndn.OnData;
+import net.named_data.jndn.OnTimeout;
+import java.util.logging.Logger;
+
+/**
+ * Provide a client to simplify information retrieval over the NDN network.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class SimpleClient implements Client {
+
+ public static final long DEFAULT_SLEEP_TIME = 20;
+ public static final long DEFAULT_TIMEOUT = 2000;
+ private static final Logger logger = Logger.getLogger(SimpleClient.class.getName());
+ private static SimpleClient defaultInstance;
+
+ /**
+ * Singleton access for simpler client use
+ *
+ * @return
+ */
+ public static SimpleClient getDefault() {
+ if (defaultInstance == null) {
+ defaultInstance = new SimpleClient();
+ }
+ return defaultInstance;
+ }
+
+ /**
+ * Asynchronously request the Data for an Interest. This will send the
+ * Interest and return immediately; use futureData.get() to block until the
+ * Data returns (see FutureData) or manage the event processing independently.
+ *
+ * @param face
+ * @param interest
+ * @return
+ */
+ @Override
+ public Future<Data> getAsync(Face face, Interest interest) {
+ final FutureData futureData = new FutureData(face, interest.getName());
+
+ // send interest
+ try {
+ face.expressInterest(interest, new OnData() {
+ @Override
+ public void onData(Interest interest, Data data) {
+ futureData.resolve(data);
+ }
+ }, new OnTimeout() {
+ @Override
+ public void onTimeout(Interest interest) {
+ futureData.reject(new TimeoutException());
+ }
+ });
+ } catch (IOException e) {
+ logger.log(Level.WARNING, "IO failure while sending interest: ", e);
+ futureData.reject(e);
+ }
+
+ return futureData;
+ }
+
+ /**
+ * Synchronously retrieve the Data for a Name using a default interest (e.g. 2
+ * second timeout); this will block until complete (i.e. either data is
+ * received or the interest times out).
+ *
+ * @param face
+ * @param name
+ * @return
+ */
+ public Future<Data> getAsync(Face face, Name name) {
+ return getAsync(face, getDefaultInterest(name));
+ }
+
+ /**
+ * Synchronously retrieve the Data for an Interest; this will block until
+ * complete (i.e. either data is received or the interest times out).
+ *
+ * @param face
+ * @param interest
+ * @return Data packet or null
+ * @throws java.io.IOException
+ */
+ @Override
+ public Data getSync(Face face, Interest interest) throws IOException {
+ try {
+ return getAsync(face, interest).get();
+ } catch (ExecutionException | InterruptedException e) {
+ logger.log(Level.WARNING, "Failed to retrieve data.", e);
+ throw new IOException("Failed to retrieve data.", e);
+ }
+ }
+
+ /**
+ * Synchronously retrieve the Data for a Name using a default interest (e.g. 2
+ * second timeout); this will block until complete (i.e. either data is
+ * received or the interest times out).
+ *
+ * @param face
+ * @param name
+ * @return
+ * @throws java.io.IOException
+ */
+ public Data getSync(Face face, Name name) throws IOException {
+ return getSync(face, getDefaultInterest(name));
+ }
+
+ /**
+ * Create a default interest for a given Name using some common settings: -
+ * lifetime: 2 seconds
+ *
+ * @param name
+ * @return
+ */
+ public static Interest getDefaultInterest(Name name) {
+ Interest interest = new Interest(name, DEFAULT_TIMEOUT);
+ return interest;
+ }
+}
diff --git a/src/main/java/com/intel/jndn/utils/FutureData.java b/src/main/java/com/intel/jndn/utils/client/FutureData.java
similarity index 83%
rename from src/main/java/com/intel/jndn/utils/FutureData.java
rename to src/main/java/com/intel/jndn/utils/client/FutureData.java
index 0b4e24d..dc6b393 100644
--- a/src/main/java/com/intel/jndn/utils/FutureData.java
+++ b/src/main/java/com/intel/jndn/utils/client/FutureData.java
@@ -11,7 +11,7 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils;
+package com.intel.jndn.utils.client;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
@@ -24,23 +24,23 @@
import net.named_data.jndn.encoding.EncodingException;
/**
- * Reference to a Packet that has yet to be returned from the network; see use
- * in WindowBuffer.java. Usage:
+ * Reference to a Packet that has yet to be returned from the network. Usage:
+ *
* <pre><code>
- * FuturePacket futurePacket = new FuturePacket(face);
+ * FutureData futureData = new FutureData(face, interest.getName());
* face.expressInterest(interest, new OnData(){
- * ... futurePacket.resolve(data); ...
+ * ... futureData.resolve(data); ...
* }, new OnTimeout(){
- * ... futurePacket.reject(new TimeoutException());
+ * ... futureData.reject(new TimeoutException());
* });
- * Packet resolvedPacket = futurePacket.get(); // will block and call face.processEvents() until complete
+ * Data resolvedData = futureData.get(); // will block and call face.processEvents() until complete
* </code></pre>
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
public class FutureData implements Future<Data> {
- private final Face face;
+ protected final Face face;
private final Name name;
private Data data;
private boolean cancelled = false;
@@ -58,7 +58,7 @@
}
/**
- * Get the packet interest name
+ * Get the Interest name.
*
* @return
*/
@@ -95,7 +95,7 @@
*/
@Override
public boolean isDone() {
- return data != null || error != null;
+ return data != null || error != null || isCancelled();
}
/**
@@ -125,7 +125,8 @@
*/
@Override
public Data get() throws InterruptedException, ExecutionException {
- while (!isDone() && !isCancelled()) {
+ while (!isDone()) {
+ // process face events
try {
synchronized (face) {
face.processEvents();
@@ -134,15 +135,17 @@
throw new ExecutionException("Failed to retrieve packet.", e);
}
}
+
// case: cancelled
if (cancelled) {
throw new InterruptedException("Interrupted by user.");
}
+
// case: error
if (error != null) {
throw new ExecutionException("Future rejected with error.", error);
}
- // case: packet
+
return data;
}
@@ -160,7 +163,9 @@
public Data get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
long interval = TimeUnit.MILLISECONDS.convert(timeout, unit);
long endTime = System.currentTimeMillis() + interval;
- while (!isDone() && !isCancelled() && System.currentTimeMillis() < endTime) {
+ long currentTime = System.currentTimeMillis();
+ while (!isDone() && !isCancelled() && currentTime <= endTime) {
+ // process face events
try {
synchronized (face) {
face.processEvents();
@@ -168,20 +173,25 @@
} catch (EncodingException | IOException e) {
throw new ExecutionException("Failed to retrieve packet.", e);
}
+
+ currentTime = System.currentTimeMillis();
}
- // case: timed out
- if (System.currentTimeMillis() < endTime) {
- throw new TimeoutException("Timed out");
- }
+
// case: cancelled
if (cancelled) {
throw new InterruptedException("Interrupted by user.");
}
+
// case: error
if (error != null) {
throw new ExecutionException("Future rejected with error.", error);
}
- // case: packet
+
+ // case: timed out
+ if (currentTime > endTime) {
+ throw new TimeoutException("Timed out.");
+ }
+
return data;
}
}
diff --git a/src/main/java/com/intel/jndn/utils/client/SegmentedFutureData.java b/src/main/java/com/intel/jndn/utils/client/SegmentedFutureData.java
new file mode 100644
index 0000000..dec33df
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/SegmentedFutureData.java
@@ -0,0 +1,172 @@
+/*
+ * jndn-utils
+ * Copyright (c) 2015, Intel Corporation.
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms and conditions of the GNU Lesser General Public License,
+ * version 3, as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope it will be useful, but WITHOUT ANY
+ * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
+ * more details.
+ */
+package com.intel.jndn.utils.client;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import net.named_data.jndn.Data;
+import net.named_data.jndn.Name;
+import net.named_data.jndn.util.Blob;
+
+/**
+ * Represents a list of Packets that have been requested asynchronously and have
+ * yet to be returned from the network. Usage:
+ *
+ * <pre><code>
+ * SegmentedFutureData segmentedFutureData = new SegmentedFutureData(face, name, futureDataList);
+ * Data data = segmentedFutureData.get(); // will block until complete
+ * </code></pre>
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class SegmentedFutureData implements Future<Data> {
+
+ private final Name name;
+ List<Future<Data>> segments;
+ private boolean cancelled = false;
+
+ /**
+ * Constructor
+ *
+ * @param name this will be the name of the returned Data packet, regardless
+ * of suffixes (e.g. segment components) on each segment packet
+ * @param segments
+ */
+ public SegmentedFutureData(Name name, List<Future<Data>> segments) {
+ this.name = name;
+ this.segments = segments;
+ }
+
+ /**
+ * Get the Interest name; this will also be the name of the Data packet
+ * returned from get().
+ *
+ * @return
+ */
+ public Name getName() {
+ return name;
+ }
+
+ /**
+ * Cancel the current request.
+ *
+ * @param mayInterruptIfRunning
+ * @return
+ */
+ @Override
+ public boolean cancel(boolean mayInterruptIfRunning) {
+ cancelled = true;
+ return cancelled;
+ }
+
+ /**
+ * Determine if this request is cancelled.
+ *
+ * @return
+ */
+ @Override
+ public boolean isCancelled() {
+ return cancelled;
+ }
+
+ /**
+ * Determine if the request has completed (successfully or not).
+ *
+ * @return
+ */
+ @Override
+ public boolean isDone() {
+ // check for errors, cancellation
+ if (isCancelled()) {
+ return true;
+ }
+
+ // check each segment for completion
+ for (Future<Data> futureData : segments) {
+ if (!futureData.isDone()) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+
+ /**
+ * Block until packet is retrieved.
+ *
+ * @return
+ * @throws InterruptedException
+ * @throws ExecutionException
+ */
+ @Override
+ public Data get() throws InterruptedException, ExecutionException {
+ // aggregate bytes
+ ByteArrayOutputStream content = new ByteArrayOutputStream();
+ for (Future<Data> futureData : segments) {
+ try {
+ content.write(futureData.get().getContent().getImmutableArray());
+ } catch (ExecutionException | IOException | InterruptedException e) {
+ throw new ExecutionException("Failed while aggregating retrieved packets.", e);
+ }
+ }
+
+ // build aggregated packet (copy first packet)
+ Data data = new Data(segments.get(0).get());
+ data.setName(getName());
+ data.setContent(new Blob(content.toByteArray()));
+ return data;
+ }
+
+ /**
+ * Block until packet is retrieved or timeout is reached.
+ *
+ * @param timeout
+ * @param unit
+ * @return
+ * @throws InterruptedException
+ * @throws ExecutionException
+ * @throws TimeoutException
+ */
+ @Override
+ public Data get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+ long interval = TimeUnit.MILLISECONDS.convert(timeout, unit);
+ long endTime = System.currentTimeMillis() + interval;
+
+ // aggregate bytes
+ ByteArrayOutputStream content = new ByteArrayOutputStream();
+ for (Future<Data> futureData : segments) {
+ try {
+ content.write(futureData.get().getContent().getImmutableArray());
+ } catch (ExecutionException | IOException | InterruptedException e) {
+ throw new ExecutionException("Failed while aggregating retrieved packets.", e);
+ }
+
+ // check for timeout
+ if (System.currentTimeMillis() > endTime) {
+ throw new TimeoutException("Timed out.");
+ }
+ }
+
+ // build aggregated packet (copy first packet)
+ Data data = new Data(segments.get(0).get());
+ data.setName(getName());
+ data.setContent(new Blob(content.toByteArray()));
+ return data;
+ }
+}
diff --git a/src/main/java/com/intel/jndn/utils/server/SegmentedServerHelper.java b/src/main/java/com/intel/jndn/utils/server/SegmentedServerHelper.java
new file mode 100644
index 0000000..ce607cf
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/server/SegmentedServerHelper.java
@@ -0,0 +1,97 @@
+/*
+ * jndn-utils
+ * Copyright (c) 2015, Intel Corporation.
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms and conditions of the GNU Lesser General Public License,
+ * version 3, as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope it will be useful, but WITHOUT ANY
+ * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
+ * more details.
+ */
+package com.intel.jndn.utils.server;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import net.named_data.jndn.Data;
+import net.named_data.jndn.Name;
+import net.named_data.jndn.util.Blob;
+
+/**
+ * Helper for segmenting an input stream into a list of Data packets.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class SegmentedServerHelper {
+
+ public static final int DEFAULT_SEGMENT_SIZE = 4096;
+
+ /**
+ * Segment a stream of bytes into a list of Data packets; this must read all
+ * the bytes first in order to determine the end segment for FinalBlockId.
+ *
+ * @param template
+ * @param bytes
+ * @return
+ * @throws IOException
+ */
+ public static List<Data> segment(Data template, InputStream bytes) throws IOException {
+ return segment(template, bytes, DEFAULT_SEGMENT_SIZE);
+ }
+
+ /**
+ * Segment a stream of bytes into a list of Data packets; this must read all
+ * the bytes first in order to determine the end segment for FinalBlockId.
+ *
+ * @param template
+ * @param bytes
+ * @param segmentSize
+ * @return
+ * @throws IOException
+ */
+ public static List<Data> segment(Data template, InputStream bytes, int segmentSize) throws IOException {
+ List<Data> segments = new ArrayList<>();
+ byte[] buffer_ = readAll(bytes);
+ int numBytes = buffer_.length;
+ int numPackets = (int) Math.ceil((double) numBytes / segmentSize);
+ ByteBuffer buffer = ByteBuffer.wrap(buffer_, 0, numBytes);
+ Name.Component lastSegment = Name.Component.fromNumberWithMarker(numPackets - 1, 0x00);
+
+ for (int i = 0; i < numPackets; i++) {
+ Data segment = new Data(template);
+ segment.getName().appendSegment(i);
+ segment.getMetaInfo().setFinalBlockId(lastSegment);
+ byte[] content = new byte[Math.min(segmentSize, buffer.remaining())];
+ buffer.get(content);
+ segment.setContent(new Blob(content));
+ segments.add(segment);
+ }
+
+ return segments;
+ }
+
+ /**
+ * Read all the bytes in an input stream.
+ *
+ * @param bytes
+ * @return
+ * @throws IOException
+ */
+ public static byte[] readAll(InputStream bytes) throws IOException {
+ ByteArrayOutputStream builder = new ByteArrayOutputStream();
+ int read = bytes.read();
+ while (read != -1) {
+ builder.write(read);
+ read = bytes.read();
+ }
+ builder.flush();
+ bytes.close();
+ return builder.toByteArray();
+ }
+}
diff --git a/src/main/resources/log4j2-test.xml b/src/main/resources/log4j2-test.xml
deleted file mode 100644
index badfbef..0000000
--- a/src/main/resources/log4j2-test.xml
+++ /dev/null
@@ -1,13 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<Configuration status="WARN">
- <Appenders>
- <Console name="Console" target="SYSTEM_OUT">
- <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
- </Console>
- </Appenders>
- <Loggers>
- <Root level="trace">
- <AppenderRef ref="Console"/>
- </Root>
- </Loggers>
-</Configuration>
\ No newline at end of file
diff --git a/src/main/resources/log4j2.xml b/src/main/resources/log4j2.xml
deleted file mode 100644
index 66435f2..0000000
--- a/src/main/resources/log4j2.xml
+++ /dev/null
@@ -1,13 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<Configuration status="WARN">
- <Appenders>
- <Console name="Console" target="SYSTEM_OUT">
- <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
- </Console>
- </Appenders>
- <Loggers>
- <Root level="warn">
- <AppenderRef ref="Console"/>
- </Root>
- </Loggers>
-</Configuration>
\ No newline at end of file
diff --git a/src/test/java/com/intel/jndn/utils/ClientTest.java b/src/test/java/com/intel/jndn/utils/ClientTest.java
index d4ca951..af22560 100644
--- a/src/test/java/com/intel/jndn/utils/ClientTest.java
+++ b/src/test/java/com/intel/jndn/utils/ClientTest.java
@@ -16,7 +16,9 @@
import org.junit.Test;
import static org.junit.Assert.*;
import com.intel.jndn.mock.MockTransport;
+import java.io.IOException;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.Logger;
@@ -28,20 +30,21 @@
/**
* Test Client.java
+ *
* @author Andrew Brown <andrew.brown@intel.com>
*/
public class ClientTest {
- /**
- * Setup logging
- */
- private static final Logger logger = Logger.getLogger(Client.class.getName());
+ private static final Logger logger = Logger.getLogger(SimpleClient.class.getName());
+ public ExpectedException thrown = ExpectedException.none();
/**
* Test retrieving data synchronously
+ *
+ * @throws java.io.IOException
*/
@Test
- public void testGetSync() {
+ public void testGetSync() throws IOException {
// setup face
MockTransport transport = new MockTransport();
Face face = new Face(transport, null);
@@ -53,7 +56,7 @@
// retrieve data
logger.info("Client expressing interest synchronously: /test/sync");
- Client client = new Client();
+ SimpleClient client = new SimpleClient();
Data data = client.getSync(face, new Name("/test/sync"));
assertEquals(new Blob("...").buf(), data.getContent().buf());
}
@@ -76,9 +79,9 @@
// retrieve data
logger.info("Client expressing interest asynchronously: /test/async");
- Client client = new Client();
- FutureData futureData = client.getAsync(face, new Name("/test/async"));
-
+ SimpleClient client = new SimpleClient();
+ Future<Data> futureData = client.getAsync(face, new Name("/test/async"));
+
assertTrue(!futureData.isDone());
futureData.get();
assertTrue(futureData.isDone());
@@ -87,10 +90,10 @@
/**
* Test that asynchronous client times out correctly
- *
- * @throws InterruptedException
+ *
+ * @throws InterruptedException
*/
- @Test
+ @Test(expected = TimeoutException.class)
public void testTimeout() throws InterruptedException, ExecutionException, TimeoutException {
// setup face
MockTransport transport = new MockTransport();
@@ -98,9 +101,9 @@
// retrieve non-existent data, should timeout
logger.info("Client expressing interest asynchronously: /test/timeout");
- FutureData futureData = Client.getDefault().getAsync(face, new Name("/test/timeout"));
-
- ExpectedException.none().expect(TimeoutException.class);
+ Future<Data> futureData = SimpleClient.getDefault().getAsync(face, new Name("/test/timeout"));
+
+ // expect an exception
futureData.get(50, TimeUnit.MILLISECONDS);
}
}
diff --git a/src/test/java/com/intel/jndn/utils/SegmentedClientTest.java b/src/test/java/com/intel/jndn/utils/SegmentedClientTest.java
index dd75f3e..e6773ec 100644
--- a/src/test/java/com/intel/jndn/utils/SegmentedClientTest.java
+++ b/src/test/java/com/intel/jndn/utils/SegmentedClientTest.java
@@ -14,8 +14,14 @@
package com.intel.jndn.utils;
import com.intel.jndn.mock.MockFace;
+import com.intel.jndn.mock.MockTransport;
import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.logging.Logger;
import net.named_data.jndn.Data;
+import net.named_data.jndn.Face;
import net.named_data.jndn.Interest;
import net.named_data.jndn.Name;
import net.named_data.jndn.Name.Component;
@@ -32,8 +38,12 @@
*/
public class SegmentedClientTest {
+ private static final Logger logger = Logger.getLogger(SimpleClient.class.getName());
+
/**
* Test of getSync method, of class SegmentedClient.
+ *
+ * @throws java.lang.Exception
*/
@Test
public void testGetSync() throws Exception {
@@ -61,4 +71,27 @@
Data data = SegmentedClient.getDefault().getSync(face, new Name("/segmented/data").appendSegment(0));
assertEquals(10, data.getContent().size());
}
+
+ /**
+ * Test that a failed request fails with an exception.
+ *
+ * @throws java.lang.Exception
+ */
+ @Test(expected = ExecutionException.class)
+ public void testFailureToRetrieve() throws Exception {
+ // setup face
+ MockTransport transport = new MockTransport();
+ Face face = new Face(transport, null);
+
+ // retrieve non-existent data, should timeout
+ logger.info("Client expressing interest asynchronously: /test/no-data");
+ List<Future<Data>> futureSegments = SegmentedClient.getDefault().getAsyncList(face, new Name("/test/no-data"));
+
+ // the list of future packets should be initialized
+ assertEquals(1, futureSegments.size());
+ assertTrue(futureSegments.get(0).isDone());
+
+ // should throw error
+ futureSegments.get(0).get();
+ }
}
diff --git a/src/test/java/com/intel/jndn/utils/ServerTest.java b/src/test/java/com/intel/jndn/utils/ServerTest.java
index 2488f0b..0c2645b 100644
--- a/src/test/java/com/intel/jndn/utils/ServerTest.java
+++ b/src/test/java/com/intel/jndn/utils/ServerTest.java
@@ -33,7 +33,7 @@
/**
* Setup logging
*/
- private static final Logger logger = Logger.getLogger(Client.class.getName());
+ private static final Logger logger = Logger.getLogger(SimpleClient.class.getName());
/**
* Test on functionality
diff --git a/src/test/java/com/intel/jndn/utils/server/SegmentedServerHelperTest.java b/src/test/java/com/intel/jndn/utils/server/SegmentedServerHelperTest.java
new file mode 100644
index 0000000..db4a8eb
--- /dev/null
+++ b/src/test/java/com/intel/jndn/utils/server/SegmentedServerHelperTest.java
@@ -0,0 +1,81 @@
+/*
+ * jndn-utils
+ * Copyright (c) 2015, Intel Corporation.
+ *
+ * This program is free software; you can redistribute it and/or modify it
+ * under the terms and conditions of the GNU Lesser General Public License,
+ * version 3, as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope it will be useful, but WITHOUT ANY
+ * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+ * FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
+ * more details.
+ */
+package com.intel.jndn.utils.server;
+
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import java.util.List;
+import net.named_data.jndn.Data;
+import net.named_data.jndn.Name;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+/**
+ * Test the SegmentedServerHelper
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class SegmentedServerHelperTest {
+
+ /**
+ * Test of segment method, of class SegmentedServerHelper.
+ *
+ * @throws java.lang.Exception
+ */
+ @Test
+ public void testSegmentation() throws Exception {
+ final Data template = new Data(new Name("/segmented/data"));
+ final InputStream content = new ByteArrayInputStream("0123456789".getBytes());
+ List<Data> segments = SegmentedServerHelper.segment(template, content, 1);
+ assertEquals(10, segments.size());
+
+ // test first packet
+ assertEquals(0, segments.get(0).getName().get(-1).toSegment());
+ assertEquals(9, segments.get(0).getMetaInfo().getFinalBlockId().toSegment());
+ assertEquals("0", segments.get(0).getContent().toString());
+
+ // test last packet
+ assertEquals(9, segments.get(9).getName().get(-1).toSegment());
+ assertEquals(9, segments.get(9).getMetaInfo().getFinalBlockId().toSegment());
+ assertEquals("9", segments.get(9).getContent().toString());
+ }
+
+ /**
+ * Test of segment method, of class SegmentedServerHelper.
+ *
+ * @throws java.lang.Exception
+ */
+ @Test
+ public void testSegmentationDifferentSizes() throws Exception {
+ final Data template = new Data(new Name("/segmented/data"));
+
+ // size 2
+ final InputStream content2 = new ByteArrayInputStream("0123456789".getBytes());
+ List<Data> segments2 = SegmentedServerHelper.segment(template, content2, 2);
+ assertEquals(5, segments2.size());
+ assertEquals("89", segments2.get(4).getContent().toString());
+
+ // size 3
+ final InputStream content3 = new ByteArrayInputStream("0123456789".getBytes());
+ List<Data> segments3 = SegmentedServerHelper.segment(template, content3, 3);
+ assertEquals(4, segments3.size());
+ assertEquals("9", segments3.get(3).getContent().toString());
+
+ // size 4
+ final InputStream content4 = new ByteArrayInputStream("0123456789".getBytes());
+ List<Data> segments4 = SegmentedServerHelper.segment(template, content4, 4);
+ assertEquals(3, segments4.size());
+ assertEquals("89", segments4.get(2).getContent().toString());
+ }
+}