Re-factor and upgrade to support Java 8 CompletableFutures as well as AdvancedClient capabilities
The squashed commits include:
Move clients to separate package; fix SimpleClient integration test
Move common test functionality to TestHelper
Add SegmentedClient integration test
Reduce logging messages; add documentation
Bump version; remove unnecessary logging configuration
doclint turns javadoc warnings into errors; ignore for now
see http://blog.joda.org/2014/02/turning-off-doclint-in-jdk-8-javadoc.html
Add snapshot repository for deploying snapshots
Bump major version due to platform change (Java 7 to Java 8)
Add sync retrieval integration test
Re-factor locations of classes: all interfaces in top level folder, change name of PipelineStage to ProcessingStage
In progress commit, tests broken
Refactor, WIP
Tweak synchronization, processing of segmented packets
Move tests around
Fix TestHelper bug
Complete tests
WIP
Add streaming test
Fix segmented server integration test
Finish refactoring
Update POM and README
Update documentation
diff --git a/src/main/java/com/intel/jndn/utils/ProcessingStage.java b/src/main/java/com/intel/jndn/utils/ProcessingStage.java
new file mode 100644
index 0000000..68b2649
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/ProcessingStage.java
@@ -0,0 +1,34 @@
+/*
+ * 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;
+
+/**
+ * Provide a generic API for processing Interest and Data packets; e.g. a data
+ * processing stage may convert a data packet with unencrypted content to one
+ * with encrypted content.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public interface ProcessingStage<T, Y> {
+
+ /**
+ * Process the input object.
+ *
+ * @param input the object to be processed
+ * @return a processed object (this may be the same instance as the input or
+ * may be a new object)
+ * @throws Exception if the processing fails
+ */
+ public Y process(T input) throws Exception;
+}
diff --git a/src/main/java/com/intel/jndn/utils/repository/Repository.java b/src/main/java/com/intel/jndn/utils/Repository.java
similarity index 94%
rename from src/main/java/com/intel/jndn/utils/repository/Repository.java
rename to src/main/java/com/intel/jndn/utils/Repository.java
index ab9a127..fcc0172 100644
--- a/src/main/java/com/intel/jndn/utils/repository/Repository.java
+++ b/src/main/java/com/intel/jndn/utils/Repository.java
@@ -11,8 +11,9 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.repository;
+package com.intel.jndn.utils;
+import com.intel.jndn.utils.repository.impl.DataNotFoundException;
import net.named_data.jndn.Data;
import net.named_data.jndn.Interest;
diff --git a/src/main/java/com/intel/jndn/utils/SegmentedClient.java b/src/main/java/com/intel/jndn/utils/SegmentedClient.java
deleted file mode 100644
index d076906..0000000
--- a/src/main/java/com/intel/jndn/utils/SegmentedClient.java
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * 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.SegmentedDataReassembler;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
-import java.util.function.Function;
-import java.util.logging.Level;
-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.encoding.EncodingException;
-
-/**
- * Provide a client to simplify retrieving segmented Data packets over the NDN
- * network. This class expects the Data producer to follow the NDN naming
- * conventions (see http://named-data.net/doc/tech-memos/naming-conventions.pdf)
- * and produce Data packets with a valid segment as the last component of their
- * name; additionally, at least the first packet should set the FinalBlockId of
- * the packet's MetaInfo (see
- * http://named-data.net/doc/ndn-tlv/data.html#finalblockid).
- *
- * @author Andrew Brown <andrew.brown@intel.com>
- */
-public class SegmentedClient extends SimpleClient {
-
- private static SegmentedClient defaultInstance;
- private static final Logger logger = Logger.getLogger(SegmentedClient.class.getName());
-
- /**
- * Singleton access for simpler client use.
- *
- * @return
- */
- public static SegmentedClient getDefault() {
- if (defaultInstance == null) {
- defaultInstance = new SegmentedClient();
- }
- return defaultInstance;
- }
-
- /**
- * See {@link SimpleClient#SimpleClient(long, long)}
- *
- * @param sleepTime
- * @param interestLifetime
- */
- public SegmentedClient(long sleepTime, long interestLifetime) {
- super(sleepTime, interestLifetime);
- }
-
- /**
- * Build a client with default parameters
- */
- public SegmentedClient() {
- super();
- }
-
- /**
- * Asynchronously send Interest packets for a segmented result; will not
- * block, but will wait for the first packet to return before sending
- * remaining interests until using the specified FinalBlockId. Will retrieve
- * non-segmented packets as well.
- *
- * @param face the {@link Face} on which to retrieve the packets
- * @param interest should include either a ChildSelector or an initial segment
- * number; the initial segment number will be cut off in the de-segmented
- * packet.
- * @return a list of FutureData packets; if the first segment fails, the list
- * will contain one FutureData with the failure exception
- */
- @Override
- public CompletableFuture<Data> getAsync(final Face face, Interest interest) {
- final long firstSegmentId = parseFirstSegmentId(interest);
-
- logger.log(Level.FINER, "Requesting segmented data packets: " + interest.getName().toUri());
- CompletableFuture<Data> allData = super.getAsync(face, interest).thenCompose(new Function<Data, CompletionStage<Data>>() {
- @Override
- public CompletionStage<Data> apply(Data firstSegment) {
- try {
- logger.log(Level.FINER, "Received first data packet: " + interest.getName().toUri());
- long lastSegmentId = parseLastSegmentId(firstSegment);
- Interest template = new Interest(interest);
- template.setName(removeSegment(firstSegment.getName()));
-
- // request subsequent segments using FinalBlockId and the Interest template
- return requestRemainingSegments(face, template, firstSegment, firstSegmentId + 1, lastSegmentId);
- } catch (EncodingException ex) {
- logger.log(Level.FINER, "No segment ID found in FinalBlockId, assuming first packet is only packet.");
- return CompletableFuture.completedFuture(firstSegment);
- }
- }
- });
-
- return allData;
- }
-
- /**
- * @param interest the request {@link Interest}
- * @return the first segment the interest is requesting, or 0 if none found
- */
- private long parseFirstSegmentId(Interest interest) {
- try {
- return interest.getName().get(-1).toSegment();
- } catch (EncodingException e) {
- 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
- }
- return 0;
- }
- }
-
- /**
- * @param firstData the first returned {@link Data}
- * @return the last segment number available as specified in the FinalBlockId
- * @throws EncodingException
- */
- private long parseLastSegmentId(Data firstData) throws EncodingException {
- return firstData.getMetaInfo().getFinalBlockId().toSegment();
- }
-
- /**
- * Send interests for remaining segments; adding them to the segmented future
- *
- * @param face
- * @param segmentedData
- * @param template
- * @param fromSegment
- * @param toSegment
- */
- private CompletableFuture<Data> requestRemainingSegments(Face face, Interest template, Data firstSegment, long fromSegment, long toSegment) {
- List<CompletableFuture<Data>> segments = new ArrayList<>();
- segments.add(CompletableFuture.completedFuture(firstSegment));
-
- for (long i = fromSegment; i <= toSegment; i++) {
- Interest segmentedInterest = new Interest(template);
- segmentedInterest.getName().appendSegment(i);
- CompletableFuture<Data> futureData = super.getAsync(face, segmentedInterest);
- segments.add(futureData);
- }
-
- return new SegmentedDataReassembler(template.getName(), segments).reassemble();
- }
-
- /**
- * Check if a name ends in a segment component; uses marker value found in the
- * NDN naming conventions (see
- * http://named-data.net/doc/tech-memos/naming-conventions.pdf).
- *
- * @param name the {@link Name} to check
- * @return true if the {@link Name} ends in a segment component
- */
- public static boolean hasSegment(Name name) {
- return name.get(-1).getValue().buf().get(0) == 0x00;
- }
-
- /**
- * @param name the {@link Name} to check
- * @return a new instance of {@link Name} with no segment component appended
- */
- public static Name removeSegment(Name name) {
- return hasSegment(name) ? name.getPrefix(-1) : new Name(name);
- }
-}
diff --git a/src/main/java/com/intel/jndn/utils/server/Server.java b/src/main/java/com/intel/jndn/utils/Server.java
similarity index 91%
rename from src/main/java/com/intel/jndn/utils/server/Server.java
rename to src/main/java/com/intel/jndn/utils/Server.java
index bcf7f3c..197a52a 100644
--- a/src/main/java/com/intel/jndn/utils/server/Server.java
+++ b/src/main/java/com/intel/jndn/utils/Server.java
@@ -11,7 +11,7 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.server;
+package com.intel.jndn.utils;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import net.named_data.jndn.Data;
@@ -47,7 +47,7 @@
* has produced a packet); also, stages should be processed in the order they
* are added.
*
- * @param pipelineStage a Data-to-Data processing stage
+ * @param stage a Data-to-Data processing stage
*/
- public void addPipelineStage(PipelineStage<Data, Data> pipelineStage);
+ public void addPostProcessingStage(ProcessingStage<Data, Data> stage);
}
diff --git a/src/main/java/com/intel/jndn/utils/client/DataStream.java b/src/main/java/com/intel/jndn/utils/client/DataStream.java
new file mode 100644
index 0000000..7f33870
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/DataStream.java
@@ -0,0 +1,75 @@
+/*
+ * 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 com.intel.jndn.utils.client.impl.StreamException;
+import net.named_data.jndn.Data;
+import net.named_data.jndn.OnData;
+import net.named_data.jndn.OnTimeout;
+
+/**
+ * Define a stream of {@link Data} packets as they are retrieved from the
+ * network and provide methods for observing stream events
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public interface DataStream extends OnData, OnTimeout, OnComplete, OnException {
+
+ /**
+ * @return true if the stream is complete
+ */
+ public boolean isComplete();
+
+ /**
+ * @return the current list of packets retrieved; this may change as more
+ * packets are added
+ */
+ public Data[] list();
+
+ /**
+ * @return an assembled packet containing the concatenated bytes of all
+ * received packets
+ * @throws StreamException if assembly fails
+ */
+ public Data assemble() throws StreamException;
+
+ /**
+ * Watch all {@link OnData} events
+ *
+ * @param onData the callback fired
+ */
+ public void observe(OnData onData);
+
+ /**
+ * Watch all {@link OnComplete} events
+ *
+ * @param onComplete the callback fired
+ */
+ public void observe(OnComplete onComplete);
+
+ /**
+ * Watch all {@link OnException} events
+ *
+ * @param onException the callback fired
+ */
+ public void observe(OnException onException);
+
+ /**
+ * Watch all {@link OnTimeout} events
+ *
+ * @param onTimeout the callback fired
+ */
+ public void observe(OnTimeout onTimeout);
+
+}
diff --git a/src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java b/src/main/java/com/intel/jndn/utils/client/OnComplete.java
similarity index 74%
copy from src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java
copy to src/main/java/com/intel/jndn/utils/client/OnComplete.java
index 732cc9c..6cbbe86 100644
--- a/src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java
+++ b/src/main/java/com/intel/jndn/utils/client/OnComplete.java
@@ -11,13 +11,17 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.repository;
+package com.intel.jndn.utils.client;
/**
- * Thrown when a {@link Repository} cannot retrieve a stored packet.
+ * Callback fired when an activity is finished.
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
-public class DataNotFoundException extends Exception {
+public interface OnComplete {
+ /**
+ * Called when an activity is finished
+ */
+ public void onComplete();
}
diff --git a/src/main/java/com/intel/jndn/utils/client/OnException.java b/src/main/java/com/intel/jndn/utils/client/OnException.java
new file mode 100644
index 0000000..d4c440d
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/OnException.java
@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+/**
+ * Callback fired when an activity throws an exception; this is necessary because
+ * some actions are performed asynchronously (e.g. in a thread pool) and exceptions
+ * could otherwise be lost.
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public interface OnException {
+
+ /**
+ * Called when an activity throws an exception
+ * @param exception the exception thrown
+ */
+ public void onException(Exception exception);
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/RetryClient.java b/src/main/java/com/intel/jndn/utils/client/RetryClient.java
new file mode 100644
index 0000000..9a5dacb
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/RetryClient.java
@@ -0,0 +1,44 @@
+/*
+ * 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.IOException;
+import net.named_data.jndn.Face;
+import net.named_data.jndn.Interest;
+import net.named_data.jndn.OnData;
+import net.named_data.jndn.OnTimeout;
+
+/**
+ * Define a client that can retry {@link Interest} packets on timeout.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public interface RetryClient {
+
+ /**
+ * Retry the given interest according to some implementation-specific
+ * strategy; the implementor must correctly call the given callbacks to pass
+ * success or failure states up to the application (e.g. if after retrying,
+ * the client gives up, the application's OnTimeout should be called). Note
+ * that an interest passed to this method may timeout long after it's lifetime
+ * due to the implementation's retry strategy.
+ *
+ * @param face the {@link Face} on which to retry requests
+ * @param interest the {@link Interest} to retry
+ * @param onData the application's success callback
+ * @param onTimeout the application's failure callback
+ * @throws IOException when the client cannot perform the necessary network IO
+ */
+ public void retry(Face face, Interest interest, OnData onData, OnTimeout onTimeout) throws IOException;
+}
diff --git a/src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java b/src/main/java/com/intel/jndn/utils/client/SegmentationType.java
similarity index 61%
copy from src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java
copy to src/main/java/com/intel/jndn/utils/client/SegmentationType.java
index 732cc9c..166d81f 100644
--- a/src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java
+++ b/src/main/java/com/intel/jndn/utils/client/SegmentationType.java
@@ -11,13 +11,26 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.repository;
+package com.intel.jndn.utils.client;
/**
- * Thrown when a {@link Repository} cannot retrieve a stored packet.
+ * Documents known partition types from
+ * http://named-data.net/doc/tech-memos/naming-conventions.pdf
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
-public class DataNotFoundException extends Exception {
+public enum SegmentationType {
+ SEGMENT((byte) 0x00),
+ BYTE_OFFSET((byte) 0xFB);
+
+ private byte marker;
+
+ SegmentationType(byte marker) {
+ this.marker = marker;
+ }
+
+ public byte value() {
+ return marker;
+ }
}
diff --git a/src/main/java/com/intel/jndn/utils/client/SegmentedClient.java b/src/main/java/com/intel/jndn/utils/client/SegmentedClient.java
new file mode 100644
index 0000000..d866675
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/SegmentedClient.java
@@ -0,0 +1,39 @@
+/*
+ * 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.IOException;
+import net.named_data.jndn.Face;
+import net.named_data.jndn.Interest;
+
+/**
+ * Define a client that can retrieve segmented packets into a
+ * {@link DataStream}.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public interface SegmentedClient {
+
+ /**
+ * Asynchronously request packets until a FinalBlockId is reached. With this
+ * method, the user is responsible for calling {@link Face#processEvents()} in
+ * order for the data to be retrieved.
+ *
+ * @param face the {@link Face} on which to retry requests
+ * @param interest the {@link Interest} to retry
+ * @return a data stream of packets returned
+ * @throws IOException if the initial request fails
+ */
+ public DataStream getSegmentsAsync(Face face, Interest interest) throws IOException;
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/SegmentedDataReassembler.java b/src/main/java/com/intel/jndn/utils/client/SegmentedDataReassembler.java
deleted file mode 100644
index 0dfa2a7..0000000
--- a/src/main/java/com/intel/jndn/utils/client/SegmentedDataReassembler.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/*
- * 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 com.intel.jndn.utils.SegmentedClient;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.function.Function;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-import net.named_data.jndn.Data;
-import net.named_data.jndn.Name;
-import net.named_data.jndn.util.Blob;
-
-/**
- *
- * @author Andrew Brown <andrew.brown@intel.com>
- */
-public class SegmentedDataReassembler {
- private static final Logger logger = Logger.getLogger(SegmentedDataReassembler.class.getName());
- private final Name interestName;
- private final List<CompletableFuture<Data>> segments;
-
- public SegmentedDataReassembler(Name interestName, List<CompletableFuture<Data>> segments) {
- this.interestName = interestName;
- this.segments = segments;
- }
-
- public CompletableFuture<Data> reassemble(){
- CompletableFuture[] segmentArray = segments.toArray(new CompletableFuture[]{});
- CompletableFuture<Void> allComplete = CompletableFuture.allOf(segmentArray);
- return allComplete.thenApply(new Function<Void, Data>() {
- @Override
- public Data apply(Void t) {
- try {
- logger.finer("Re-assembling data for request: " + interestName.toUri());
- byte[] content = aggregateBytes();
- Data data = buildAggregatePacket();
- data.setContent(new Blob(content));
- return data;
- } catch (ExecutionException | InterruptedException ex) {
- logger.log(Level.FINER, "Failed to re-assemble packet for request: " + interestName.toUri(), ex);
- throw new RuntimeException(ex);
- }
- }
- });
- }
-
- /**
- * @return the array of aggregated bytes for all of the segments retrieved
- * @throws ExecutionException
- */
- private byte[] aggregateBytes() throws 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: " + interestName.toUri(), e);
- }
- }
- return content.toByteArray();
- }
-
- /**
- * @return an aggregated {@link Data} packet with no content set
- * @throws InterruptedException
- * @throws ExecutionException
- */
- private Data buildAggregatePacket() throws InterruptedException, ExecutionException {
- if (segments.isEmpty()) {
- throw new IllegalStateException("Unable to re-assemble packets; no segments added: " + interestName.toUri());
- }
- Data firstData = segments.get(0).get();
- Data aggregatedData = new Data(firstData);
- Name shortenedName = SegmentedClient.removeSegment(firstData.getName());
- aggregatedData.setName(shortenedName);
- return aggregatedData;
- }
-}
diff --git a/src/main/java/com/intel/jndn/utils/client/StreamingClient.java b/src/main/java/com/intel/jndn/utils/client/StreamingClient.java
new file mode 100644
index 0000000..3ab5fb6
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/StreamingClient.java
@@ -0,0 +1,43 @@
+/*
+ * 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.IOException;
+import java.io.InputStream;
+import net.named_data.jndn.Face;
+import net.named_data.jndn.Interest;
+
+/**
+ * Define a client that can stream content bytes that are partitioned over
+ * multiple packets.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public interface StreamingClient {
+
+ /**
+ * Asynchronously request a packet over the network and retrieve the
+ * partitioned content bytes as an input stream
+ *
+ * @param face the {@link Face} on which to make the request; call
+ * {@link Face#processEvents()} separately to complete the request
+ * @param interest the {@link Interest} to send over the network
+ * @param partitionMarker the byte marker identifying how the data packets are
+ * partitioned (e.g. segmentation, see
+ * http://named-data.net/doc/tech-memos/naming-conventions.pdf)
+ * @return a stream of content bytes
+ * @throws IOException if the stream setup fails
+ */
+ public InputStream getStreamAsync(Face face, Interest interest, SegmentationType partitionMarker) throws IOException;
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/impl/AdvancedClient.java b/src/main/java/com/intel/jndn/utils/client/impl/AdvancedClient.java
new file mode 100644
index 0000000..be1e876
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/impl/AdvancedClient.java
@@ -0,0 +1,169 @@
+/*
+ * 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.impl;
+
+import com.intel.jndn.utils.client.OnComplete;
+import com.intel.jndn.utils.client.OnException;
+import com.intel.jndn.utils.client.RetryClient;
+import com.intel.jndn.utils.client.SegmentedClient;
+import com.intel.jndn.utils.client.DataStream;
+import com.intel.jndn.utils.client.SegmentationType;
+import com.intel.jndn.utils.client.StreamingClient;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeoutException;
+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.OnTimeout;
+
+/**
+ * Implementation of a client that can handle segmented data, retries after
+ * failed requests, and streaming of data packets.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class AdvancedClient extends SimpleClient implements SegmentedClient, StreamingClient {
+
+ private static final Logger logger = Logger.getLogger(AdvancedClient.class.getName());
+ public static final int DEFAULT_MAX_RETRIES = 3;
+ private static AdvancedClient defaultInstance;
+ private final SegmentedClient segmentedClient;
+ private final RetryClient retryClient;
+ private final StreamingClient streamingClient;
+
+ /**
+ * Singleton access for simpler client use
+ *
+ * @return a default client
+ */
+ public static AdvancedClient getDefault() {
+ if (defaultInstance == null) {
+ defaultInstance = new AdvancedClient();
+ }
+ return defaultInstance;
+ }
+
+ /**
+ * Build an advanced client
+ *
+ * @param sleepTime for synchronous processing, the time to sleep the thread
+ * between {@link Face#processEvents()}
+ * @param interestLifetime the {@link Interest} lifetime for default
+ * Interests; see
+ * {@link #getAsync(net.named_data.jndn.Face, net.named_data.jndn.Name)}
+ * @param segmentedClient the {@link SegmentedClient} to use for segmented
+ * data
+ * @param retryClient the {@link RetryClient} to use for retrying failed
+ * packets
+ * @param streamingClient the {@link StreamingClient} to use for segmented
+ * data
+ */
+ public AdvancedClient(long sleepTime, long interestLifetime, SegmentedClient segmentedClient, RetryClient retryClient, StreamingClient streamingClient) {
+ super(sleepTime, interestLifetime);
+ this.segmentedClient = segmentedClient;
+ this.retryClient = retryClient;
+ this.streamingClient = streamingClient;
+ }
+
+ /**
+ * Build an advanced client using default parameters
+ */
+ public AdvancedClient() {
+ super();
+ this.segmentedClient = new DefaultSegmentedClient();
+ this.retryClient = new DefaultRetryClient(DEFAULT_MAX_RETRIES);
+ this.streamingClient = new DefaultStreamingClient();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public CompletableFuture<Data> getAsync(Face face, Interest interest) {
+ CompletableFuture<Data> future = new CompletableFuture<>();
+
+ try {
+ DataStream stream = getSegmentsAsync(face, interest);
+
+ stream.observe(new OnException() {
+ public void onException(Exception exception) {
+ future.completeExceptionally(exception);
+ }
+ });
+
+ stream.observe(new OnComplete() {
+ public void onComplete() {
+ try {
+ future.complete(stream.assemble());
+ } catch (StreamException ex) {
+ stream.onException(ex);
+ }
+ }
+ });
+ } catch (IOException ex) {
+ future.completeExceptionally(ex);
+ }
+
+ return future;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public DataStream getSegmentsAsync(Face face, Interest interest) throws IOException {
+ DataStream stream = segmentedClient.getSegmentsAsync(face, interest);
+ stream.observe(new RetryHandler(face, stream));
+ return stream;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public InputStream getStreamAsync(Face face, Interest interest, SegmentationType partitionMarker) throws IOException {
+ return streamingClient.getStreamAsync(face, interest, partitionMarker);
+ }
+
+ /**
+ * Helper class for calling the retry client on failed requests
+ */
+ private class RetryHandler implements OnTimeout {
+ private final Face face;
+ private final DataStream stream;
+
+ private RetryHandler(Face face, DataStream stream) {
+ this.face = face;
+ this.stream = stream;
+ }
+
+ @Override
+ public void onTimeout(Interest failedInterest) {
+ logger.info("Timeout: " + failedInterest.toUri());
+ try {
+ retryClient.retry(face, failedInterest, stream, new OnTimeout() {
+ @Override
+ public void onTimeout(Interest interest) {
+ stream.onException(new TimeoutException("Interest timed out despite retries: " + interest.toUri()));
+ }
+ });
+ } catch (IOException ex) {
+ stream.onException(ex);
+ }
+ }
+ }
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/impl/DataAssembler.java b/src/main/java/com/intel/jndn/utils/client/impl/DataAssembler.java
new file mode 100644
index 0000000..c1ca4fe
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/impl/DataAssembler.java
@@ -0,0 +1,79 @@
+/*
+ * 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.impl;
+
+import com.intel.jndn.utils.client.SegmentedClient;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import net.named_data.jndn.Data;
+import net.named_data.jndn.util.Blob;
+
+/**
+ * Internal class for assembling a list of {@link Data} packets into one large
+ * data packet; this implementation will use all properties of the first packet
+ * in the list and concatenate the content bytes of all packets in order.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+class DataAssembler {
+
+ private final Data[] packets;
+ private final byte marker;
+
+ public DataAssembler(Data[] packets, byte marker) {
+ this.packets = packets;
+ this.marker = marker;
+ }
+
+ /**
+ * @return a combined packet based on the properties of the first packet and
+ * the concatenated bytes of the entire list of packets
+ */
+ Data assemble() {
+ if (packets.length == 0) {
+ throw new IllegalStateException("No packets to assemble.");
+ }
+
+ // build from first returned packet
+ Data combined = new Data(packets[0]);
+ combined.setName(SegmentationHelper.removeSegment(combined.getName(), marker));
+ try {
+ combined.setContent(assembleBlob());
+ } catch (IOException ex) {
+ throw new IllegalStateException(ex);
+ }
+
+ return combined;
+ }
+
+ /**
+ * @return the concatenated bytes
+ * @throws IOException if a stream fails
+ */
+ private Blob assembleBlob() throws IOException {
+ if (packets.length == 0) {
+ return new Blob();
+ }
+
+ if (packets.length == 1) {
+ return packets[0].getContent();
+ }
+
+ ByteArrayOutputStream stream = new ByteArrayOutputStream();
+ for (Data packet : packets) {
+ stream.write(packet.getContent().getImmutableArray());
+ }
+ return new Blob(stream.toByteArray());
+ }
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/impl/DefaultRetryClient.java b/src/main/java/com/intel/jndn/utils/client/impl/DefaultRetryClient.java
new file mode 100644
index 0000000..f984ac2
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/impl/DefaultRetryClient.java
@@ -0,0 +1,121 @@
+/*
+ * 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.impl;
+
+import com.intel.jndn.utils.client.RetryClient;
+import java.io.IOException;
+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.OnData;
+import net.named_data.jndn.OnTimeout;
+
+/**
+ * Default implementation of {@link RetryClient}; on request failure, this class
+ * immediately retries the request until a maximum number of retries is reached.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class DefaultRetryClient implements RetryClient {
+
+ private static final Logger logger = Logger.getLogger(DefaultRetryClient.class.getName());
+ private final int numRetriesAllowed;
+ private int totalRetries = 0;
+
+ public DefaultRetryClient(int numRetriesAllowed) {
+ this.numRetriesAllowed = numRetriesAllowed;
+ }
+
+ /**
+ * On timeout, retry the request until the maximum number of allowed retries
+ * is reached.
+ *
+ * @param face the {@link Face} on which to retry requests
+ * @param interest the {@link Interest} to retry
+ * @param onData the application's success callback
+ * @param onTimeout the application's failure callback
+ * @throws IOException when the client cannot perform the necessary network IO
+ */
+ @Override
+ public void retry(Face face, Interest interest, OnData onData, OnTimeout onTimeout) throws IOException {
+ RetryContext context = new RetryContext(face, interest, onData, onTimeout);
+ retryInterest(context);
+ }
+
+ /**
+ * Synchronized helper method to prevent multiple threads from mashing
+ * totalRetries
+ *
+ * @param context the current request context
+ * @throws IOException when the client cannot perform the necessary network IO
+ */
+ synchronized private void retryInterest(RetryContext context) throws IOException {
+ logger.info("Retrying interest: " + context.interest.toUri());
+ context.face.expressInterest(context.interest, context, context);
+ totalRetries++;
+ }
+
+ /**
+ * @return the total number of retries logged by this client
+ */
+ public int totalRetries() {
+ return totalRetries;
+ }
+
+ /**
+ * Helper class to separate out each request context; this allows the client
+ * to accept multiple concurrent retry operations
+ */
+ private class RetryContext implements OnData, OnTimeout {
+
+ public int numFailures = 0;
+ public final Face face;
+ public final Interest interest;
+ public final OnData applicationOnData;
+ public final OnTimeout applicationOnTimeout;
+
+ public RetryContext(Face face, Interest interest, OnData applicationOnData, OnTimeout applicationOnTimeout) {
+ this.face = face;
+ this.interest = interest;
+ this.applicationOnData = applicationOnData;
+ this.applicationOnTimeout = applicationOnTimeout;
+ }
+
+ public boolean shouldRetry() {
+ return numFailures < numRetriesAllowed;
+ }
+
+ @Override
+ public void onData(Interest interest, Data data) {
+ applicationOnData.onData(interest, data);
+ }
+
+ @Override
+ public void onTimeout(Interest interest) {
+ numFailures++;
+ logger.finest("Request failed, count " + numFailures + ": " + interest.toUri());
+
+ if (shouldRetry()) {
+ try {
+ retryInterest(this);
+ } catch (IOException ex) {
+ applicationOnTimeout.onTimeout(interest);
+ }
+ } else {
+ applicationOnTimeout.onTimeout(interest);
+ }
+ }
+ }
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/impl/DefaultSegmentedClient.java b/src/main/java/com/intel/jndn/utils/client/impl/DefaultSegmentedClient.java
new file mode 100644
index 0000000..83e0baa
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/impl/DefaultSegmentedClient.java
@@ -0,0 +1,149 @@
+/*
+ * 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.impl;
+
+import com.intel.jndn.utils.client.SegmentedClient;
+import com.intel.jndn.utils.client.DataStream;
+import java.io.IOException;
+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;
+import net.named_data.jndn.OnData;
+
+/**
+ * Retrieve segments one by one until the FinalBlockId indicates an end segment;
+ * then request remaining packets. This class currently only handles segmented
+ * data (not yet byte-offset segmented data).
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class DefaultSegmentedClient implements SegmentedClient {
+
+ private static final Logger logger = Logger.getLogger(DefaultSegmentedClient.class.getName());
+ private static DefaultSegmentedClient defaultInstance;
+ private byte marker = 0x00;
+
+ /**
+ * Singleton access for simpler client use
+ *
+ * @return a default client
+ */
+ public static DefaultSegmentedClient getDefault() {
+ if (defaultInstance == null) {
+ defaultInstance = new DefaultSegmentedClient();
+ }
+ return defaultInstance;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public DataStream getSegmentsAsync(Face face, Interest interest) throws IOException {
+ SegmentedDataStream stream = new SegmentedDataStream();
+
+ // once more packets are received, request more
+ stream.observe(new SegmentationContext(stream, face));
+
+ // request first packet
+ logger.info("Interest requested: " + interest.toUri());
+ face.expressInterest(interest, stream, stream);
+
+ return stream;
+ }
+
+ /**
+ * Helper class to track the last requested segment for a given request
+ * context and request follow-on packets
+ */
+ private class SegmentationContext implements OnData {
+
+ private final SegmentedDataStream stream;
+ private final Face face;
+ private long lastRequestedSegment;
+
+ public SegmentationContext(SegmentedDataStream stream, Face face) {
+ this.stream = stream;
+ this.face = face;
+ }
+
+ private synchronized void setLastRequestedSegment(long segmentNumber) {
+ lastRequestedSegment = segmentNumber;
+ }
+
+ @Override
+ public void onData(Interest interest, Data data) {
+ try {
+ if (stream.current() >= lastRequestedSegment) {
+ Interest dataBasedName = new Interest(interest).setName(data.getName());
+ if (stream.hasEnd()) {
+ if (stream.current() < stream.end()) {
+ requestRemainingSegments(face, dataBasedName, stream);
+ }
+ } else {
+ requestNext(face, dataBasedName, stream);
+ }
+ }
+ } catch (IOException e) {
+ stream.onException(e);
+ }
+ }
+
+ private void requestRemainingSegments(Face face, Interest interest, SegmentedDataStream stream) throws IOException {
+ long from = stream.current() + 1;
+ long to = stream.end();
+ logger.info("Requesting remaining segments: from #" + from + " to #" + to);
+
+ for (long segmentNumber = stream.current() + 1; segmentNumber <= stream.end(); segmentNumber++) {
+ request(face, interest, stream, segmentNumber, marker);
+ }
+ }
+
+ private void requestNext(Face face, Interest interest, SegmentedDataStream stream) throws IOException {
+ long segmentNumber = stream.current() + 1;
+ request(face, interest, stream, segmentNumber, marker);
+ }
+
+ private void request(Face face, Interest interest, DataStream stream, long segmentNumber, byte marker) throws IOException {
+ Interest copiedInterest = replaceFinalComponent(interest, segmentNumber, marker);
+ face.expressInterest(copiedInterest, stream, stream);
+ logger.info("Interest sent: " + copiedInterest.toUri());
+ setLastRequestedSegment(segmentNumber);
+ }
+ }
+
+ /**
+ * Replace the final component of an interest name with a segmented component;
+ * if the interest name does not have a segmented component, this will add
+ * one.
+ *
+ * @param interest the request
+ * @param segmentNumber a segment number
+ * @param marker a marker to use for segmenting the packet
+ * @return a segmented interest (a copy of the passed interest)
+ */
+ protected Interest replaceFinalComponent(Interest interest, long segmentNumber, byte marker) {
+ Interest copied = new Interest(interest);
+ Component lastComponent = Component.fromNumberWithMarker(segmentNumber, marker);
+ Name newName = (SegmentationHelper.isSegmented(copied.getName(), marker))
+ ? copied.getName().getPrefix(-1)
+ : new Name(copied.getName());
+ copied.setName(newName.append(lastComponent));
+ return copied;
+ }
+
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/impl/DefaultStreamingClient.java b/src/main/java/com/intel/jndn/utils/client/impl/DefaultStreamingClient.java
new file mode 100644
index 0000000..55fa0f0
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/impl/DefaultStreamingClient.java
@@ -0,0 +1,103 @@
+/*
+ * 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.impl;
+
+import com.intel.jndn.utils.client.DataStream;
+import com.intel.jndn.utils.client.OnException;
+import com.intel.jndn.utils.client.SegmentationType;
+import com.intel.jndn.utils.client.SegmentedClient;
+import com.intel.jndn.utils.client.StreamingClient;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.util.logging.Level;
+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.OnData;
+
+/**
+ * Default implementation of {@link StreamingClient}; uses a segmented client to
+ * retrieve packets asynchronously and pipes them to a stream as they are
+ * received.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class DefaultStreamingClient implements StreamingClient {
+
+ private static final Logger logger = Logger.getLogger(DefaultStreamingClient.class.getName());
+ private SegmentedClient client;
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public InputStream getStreamAsync(Face face, Interest interest, SegmentationType partitionMarker) throws IOException {
+ return getStreamAsync(face, interest, partitionMarker, new DefaultOnException());
+ }
+
+ /**
+ *
+ * @param face
+ * @param interest
+ * @param partitionMarker
+ * @param onException
+ * @return
+ * @throws IOException
+ */
+ public InputStream getStreamAsync(Face face, Interest interest, SegmentationType partitionMarker, OnException onException) throws IOException {
+ SegmentedClient client = DefaultSegmentedClient.getDefault();
+ return getStreamAsync(client.getSegmentsAsync(face, interest), onException);
+ }
+
+ /**
+ *
+ * @param onDataStream
+ * @param onException
+ * @return
+ * @throws IOException
+ */
+ public InputStream getStreamAsync(DataStream onDataStream, OnException onException) throws IOException {
+ PipedInputStream in = new PipedInputStream();
+ PipedOutputStream out = new PipedOutputStream(in);
+
+ onDataStream.observe(onException);
+
+ onDataStream.observe(new OnData() {
+ @Override
+ public void onData(Interest interest, Data data) {
+ try {
+ out.write(data.getContent().getImmutableArray());
+ } catch (IOException ex) {
+ onDataStream.onException(ex);
+ }
+ }
+ });
+
+ return in;
+ }
+
+ /**
+ * Helper for logging exceptions
+ */
+ private class DefaultOnException implements OnException {
+
+ @Override
+ public void onException(Exception exception) {
+ logger.log(Level.SEVERE, "Streaming failed", exception);
+ }
+ }
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/impl/SegmentationHelper.java b/src/main/java/com/intel/jndn/utils/client/impl/SegmentationHelper.java
new file mode 100644
index 0000000..7a051ee
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/impl/SegmentationHelper.java
@@ -0,0 +1,66 @@
+/*
+ * 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.impl;
+
+import net.named_data.jndn.Name;
+import net.named_data.jndn.encoding.EncodingException;
+
+/**
+ * Helper methods for dealing with segmented packets (see
+ * http://named-data.net/doc/tech-memos/naming-conventions.pdf).
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class SegmentationHelper {
+
+ /**
+ * Determine if a name is segmented, i.e. if it ends with the correct marker
+ * type.
+ *
+ * @param name the name of a packet
+ * @param marker the marker type (the initial byte of the component)
+ * @return true if the name is segmented
+ */
+ public static boolean isSegmented(Name name, byte marker) {
+ return name.size() > 0 ? name.get(-1).getValue().buf().get(0) == marker : false;
+ }
+
+ /**
+ * Retrieve the segment number from the last component of a name.
+ *
+ * @param name the name of a packet
+ * @param marker the marker type (the initial byte of the component)
+ * @return the segment number
+ * @throws EncodingException if the name does not have a final component of
+ * the correct marker type
+ */
+ public static long parseSegment(Name name, byte marker) throws EncodingException {
+ if (name.size() == 0) {
+ throw new EncodingException("No components to parse.");
+ }
+ return name.get(-1).toNumberWithMarker(marker);
+ }
+
+ /**
+ * Remove a segment component from the end of a name
+ *
+ * @param name the name of a packet
+ * @param marker the marker type (the initial byte of the component)
+ * @return the new name with the segment component removed or a copy of the
+ * name if no segment component was present
+ */
+ public static Name removeSegment(Name name, byte marker) {
+ return isSegmented(name, marker) ? name.getPrefix(-1) : new Name(name);
+ }
+}
diff --git a/src/main/java/com/intel/jndn/utils/client/impl/SegmentedDataStream.java b/src/main/java/com/intel/jndn/utils/client/impl/SegmentedDataStream.java
new file mode 100644
index 0000000..41fd86a
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/client/impl/SegmentedDataStream.java
@@ -0,0 +1,199 @@
+/*
+ * 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.impl;
+
+import com.intel.jndn.utils.client.OnComplete;
+import com.intel.jndn.utils.client.OnException;
+import com.intel.jndn.utils.client.DataStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.logging.Logger;
+import java.util.stream.Collectors;
+import net.named_data.jndn.Data;
+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 net.named_data.jndn.encoding.EncodingException;
+
+/**
+ * As packets are received, they are mapped by their last component's segment
+ * marker; if the last component is not parseable, an exception is thrown and
+ * processing completes. The exception to this is if the first packet returned
+ * does not have a segment marker as the last component; in this case, the
+ * packet is assumed to be the only packet returned and is placed as the first
+ * and only packet to assemble. Observers may register callbacks to watch when
+ * data is received; if data is received out of order, the callbacks will not be
+ * fired until adjoining packets are received.
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class SegmentedDataStream implements DataStream {
+
+ private static final Logger logger = Logger.getLogger(SegmentedDataStream.class.getName());
+ private final byte PARTITION_MARKER = 0x00;
+ private volatile long current = -1;
+ private volatile long end = Long.MAX_VALUE;
+ private Map<Long, Data> packets = new HashMap<>();
+ private List<Object> observers = new ArrayList<>();
+ private Exception exception;
+
+ @Override
+ public boolean isComplete() {
+ return current == end || isCompletedExceptionally();
+ }
+
+ public boolean isCompletedExceptionally() {
+ return exception != null;
+ }
+
+ public boolean hasEnd() {
+ return end != Long.MAX_VALUE;
+ }
+
+ public long end() {
+ return end;
+ }
+
+ public long current() {
+ return current;
+ }
+
+ @Override
+ public Data[] list() {
+ return packets.values().toArray(new Data[]{});
+ }
+
+ @Override
+ public Data assemble() throws StreamException {
+ if (isCompletedExceptionally()) {
+ throw new StreamException(exception);
+ }
+
+ Object o = new LinkedList();
+ return new DataAssembler(list(), PARTITION_MARKER).assemble();
+ }
+
+ @Override
+ public void observe(OnData onData) {
+ observers.add(onData);
+ }
+
+ @Override
+ public void observe(OnComplete onComplete) {
+ observers.add(onComplete);
+ }
+
+ @Override
+ public void observe(OnException onException) {
+ observers.add(onException);
+ }
+
+ @Override
+ public void observe(OnTimeout onTimeout) {
+ observers.add(onTimeout);
+ }
+
+ @Override
+ public synchronized void onData(Interest interest, Data data) {
+ logger.info("Data received: " + data.getName().toUri());
+ long id;
+
+ // no segment component
+ if (!SegmentationHelper.isSegmented(data.getName(), PARTITION_MARKER) && packets.size() == 0) {
+ id = 0;
+ packets.put(id, data);
+
+ // mark processing complete if the first packet has no segment component
+ end = 0;
+ } // with segment component
+ else {
+ Name.Component lastComponent = data.getName().get(-1);
+ try {
+ id = lastComponent.toNumberWithMarker(PARTITION_MARKER);
+ packets.put(id, data);
+ } catch (EncodingException ex) {
+ onException(ex);
+ return;
+ }
+ }
+
+ if (hasFinalBlockId(data)) {
+ try {
+ end = data.getMetaInfo().getFinalBlockId().toNumberWithMarker(PARTITION_MARKER);
+ } catch (EncodingException ex) {
+ onException(ex);
+ }
+ }
+
+ // call data observers
+ if (isNextPacket(id)) {
+ do {
+ current++;
+ assert (packets.containsKey(current));
+ Data retrieved = packets.get(current);
+ observersOfType(OnData.class).stream().forEach((OnData cb) -> {
+ cb.onData(interest, retrieved);
+ });
+ } while (hasNextPacket());
+ }
+
+ // call completion observers
+ if (isComplete()) {
+ onComplete();
+ }
+ }
+
+ private boolean hasFinalBlockId(Data data) {
+ return data.getMetaInfo().getFinalBlockId().getValue().size() > 0;
+ }
+
+ private boolean isNextPacket(long id) {
+ return current + 1 == id;
+ }
+
+ private boolean hasNextPacket() {
+ return packets.containsKey(current + 1);
+ }
+
+ @Override
+ public synchronized void onComplete() {
+ observersOfType(OnComplete.class).stream().forEach((OnComplete cb) -> {
+ cb.onComplete();
+ });
+ }
+
+ @Override
+ public synchronized void onTimeout(Interest interest) {
+ observersOfType(OnTimeout.class).stream().forEach((OnTimeout cb) -> {
+ cb.onTimeout(interest);
+ });
+ }
+
+ @Override
+ public synchronized void onException(Exception exception) {
+ this.exception = exception;
+
+ observersOfType(OnException.class).stream().forEach((OnException cb) -> {
+ cb.onException(exception);
+ });
+ }
+
+ private <T> List<T> observersOfType(Class<T> type) {
+ return observers.stream().filter((Object o) -> type.isAssignableFrom(o.getClass())).map((o) -> (T) o).collect(Collectors.toList());
+ }
+}
diff --git a/src/main/java/com/intel/jndn/utils/SimpleClient.java b/src/main/java/com/intel/jndn/utils/client/impl/SimpleClient.java
similarity index 90%
rename from src/main/java/com/intel/jndn/utils/SimpleClient.java
rename to src/main/java/com/intel/jndn/utils/client/impl/SimpleClient.java
index 1bf8519..9e2e594 100644
--- a/src/main/java/com/intel/jndn/utils/SimpleClient.java
+++ b/src/main/java/com/intel/jndn/utils/client/impl/SimpleClient.java
@@ -11,8 +11,9 @@
* 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.impl;
+import com.intel.jndn.utils.Client;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
@@ -87,6 +88,7 @@
face.expressInterest(interest, new OnData() {
@Override
public void onData(Interest interest, Data data) {
+ logger.log(Level.FINER, "Retrieved data: " + data.getName().toUri());
futureData.complete(data);
}
}, new OnTimeout() {
@@ -118,18 +120,18 @@
@Override
public Data getSync(Face face, Interest interest) throws IOException {
CompletableFuture<Data> future = getAsync(face, interest);
+
try {
// process events until complete
while (!future.isDone()) {
synchronized (face) {
face.processEvents();
}
-
+
if (sleepTime > 0) {
- try{
+ try {
Thread.sleep(sleepTime);
- }
- catch(InterruptedException e){
+ } catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
@@ -150,11 +152,11 @@
}
/**
- * Create a default interest for a given Name using some common settings: -
- * lifetime: 2 seconds
+ * Create a default interest for a given {@link Name} using the client's
+ * passed settings (see {@link #SimpleClient(long, long)})
*
- * @param name
- * @return
+ * @param name the {@link Name} of the data to retrieve
+ * @return a default interest for the given name
*/
public Interest getDefaultInterest(Name name) {
Interest interest = new Interest(name, interestLifetime);
diff --git a/src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java b/src/main/java/com/intel/jndn/utils/client/impl/StreamException.java
similarity index 71%
copy from src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java
copy to src/main/java/com/intel/jndn/utils/client/impl/StreamException.java
index 732cc9c..59ad55a 100644
--- a/src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java
+++ b/src/main/java/com/intel/jndn/utils/client/impl/StreamException.java
@@ -11,13 +11,20 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.repository;
+package com.intel.jndn.utils.client.impl;
/**
- * Thrown when a {@link Repository} cannot retrieve a stored packet.
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
-public class DataNotFoundException extends Exception {
+public class StreamException extends Exception {
+ public StreamException(Exception exception) {
+ super(exception);
+ }
+
+ public StreamException(String message){
+ super(message);
+ }
+
}
diff --git a/src/main/java/com/intel/jndn/utils/server/pipeline/CompressionStage.java b/src/main/java/com/intel/jndn/utils/processing/impl/CompressionStage.java
similarity index 89%
rename from src/main/java/com/intel/jndn/utils/server/pipeline/CompressionStage.java
rename to src/main/java/com/intel/jndn/utils/processing/impl/CompressionStage.java
index 687bde0..21862fa 100644
--- a/src/main/java/com/intel/jndn/utils/server/pipeline/CompressionStage.java
+++ b/src/main/java/com/intel/jndn/utils/processing/impl/CompressionStage.java
@@ -11,9 +11,9 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.server.pipeline;
+package com.intel.jndn.utils.processing.impl;
-import com.intel.jndn.utils.server.PipelineStage;
+import com.intel.jndn.utils.ProcessingStage;
import java.io.ByteArrayOutputStream;
import java.util.zip.GZIPOutputStream;
import net.named_data.jndn.Data;
@@ -24,7 +24,7 @@
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
-public class CompressionStage implements PipelineStage<Data, Data> {
+public class CompressionStage implements ProcessingStage<Data, Data> {
/**
* Compress and replace the {@link Data} content. Note: this stage will return
diff --git a/src/main/java/com/intel/jndn/utils/server/pipeline/SigningStage.java b/src/main/java/com/intel/jndn/utils/processing/impl/SigningStage.java
similarity index 92%
rename from src/main/java/com/intel/jndn/utils/server/pipeline/SigningStage.java
rename to src/main/java/com/intel/jndn/utils/processing/impl/SigningStage.java
index 5296d89..3125db8 100644
--- a/src/main/java/com/intel/jndn/utils/server/pipeline/SigningStage.java
+++ b/src/main/java/com/intel/jndn/utils/processing/impl/SigningStage.java
@@ -11,9 +11,9 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.server.pipeline;
+package com.intel.jndn.utils.processing.impl;
-import com.intel.jndn.utils.server.PipelineStage;
+import com.intel.jndn.utils.ProcessingStage;
import net.named_data.jndn.Data;
import net.named_data.jndn.Name;
import net.named_data.jndn.security.KeyChain;
@@ -25,7 +25,7 @@
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
-public class SigningStage implements PipelineStage<Data, Data> {
+public class SigningStage implements ProcessingStage<Data, Data> {
private final KeyChain keyChain;
private final Name certificateName;
diff --git a/src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java b/src/main/java/com/intel/jndn/utils/repository/impl/DataNotFoundException.java
similarity index 93%
rename from src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java
rename to src/main/java/com/intel/jndn/utils/repository/impl/DataNotFoundException.java
index 732cc9c..eba700a 100644
--- a/src/main/java/com/intel/jndn/utils/repository/DataNotFoundException.java
+++ b/src/main/java/com/intel/jndn/utils/repository/impl/DataNotFoundException.java
@@ -11,7 +11,7 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.repository;
+package com.intel.jndn.utils.repository.impl;
/**
* Thrown when a {@link Repository} cannot retrieve a stored packet.
diff --git a/src/main/java/com/intel/jndn/utils/repository/ForLoopRepository.java b/src/main/java/com/intel/jndn/utils/repository/impl/ForLoopRepository.java
similarity index 97%
rename from src/main/java/com/intel/jndn/utils/repository/ForLoopRepository.java
rename to src/main/java/com/intel/jndn/utils/repository/impl/ForLoopRepository.java
index 80fedb1..e40890d 100644
--- a/src/main/java/com/intel/jndn/utils/repository/ForLoopRepository.java
+++ b/src/main/java/com/intel/jndn/utils/repository/impl/ForLoopRepository.java
@@ -11,8 +11,9 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.repository;
+package com.intel.jndn.utils.repository.impl;
+import com.intel.jndn.utils.Repository;
import java.util.ArrayList;
import java.util.List;
import net.named_data.jndn.Data;
@@ -23,7 +24,7 @@
* Store {@link Data} packets in a linked list and iterate over the list to find
* the best match; this is a subset of the functionality provided in
* {@link net.named_data.jndn.util.MemoryContentCache} and borrows the matching
- * logic from there. Code for removing stale packets is not yet implemented.
+ * logic from there.
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
diff --git a/src/main/java/com/intel/jndn/utils/DynamicServer.java b/src/main/java/com/intel/jndn/utils/server/DynamicServer.java
similarity index 95%
rename from src/main/java/com/intel/jndn/utils/DynamicServer.java
rename to src/main/java/com/intel/jndn/utils/server/DynamicServer.java
index 8206e27..60caf35 100644
--- a/src/main/java/com/intel/jndn/utils/DynamicServer.java
+++ b/src/main/java/com/intel/jndn/utils/server/DynamicServer.java
@@ -11,10 +11,10 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils;
+package com.intel.jndn.utils.server;
+import com.intel.jndn.utils.Server;
import com.intel.jndn.utils.server.RespondWithData;
-import com.intel.jndn.utils.server.Server;
import java.io.IOException;
/**
diff --git a/src/main/java/com/intel/jndn/utils/server/PipelineStage.java b/src/main/java/com/intel/jndn/utils/server/PipelineStage.java
deleted file mode 100644
index 15d23f0..0000000
--- a/src/main/java/com/intel/jndn/utils/server/PipelineStage.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/*
- * 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;
-
-/**
- *
- * @author Andrew Brown <andrew.brown@intel.com>
- */
-public interface PipelineStage<T, Y> {
- public Y process(T context) throws Exception;
-// public void setNextStage(PipelineStage<Y, ?> nextStage);
-// public PipelineStage<Y, ?> getNextStage();
-}
diff --git a/src/main/java/com/intel/jndn/utils/RepositoryServer.java b/src/main/java/com/intel/jndn/utils/server/RepositoryServer.java
similarity index 94%
rename from src/main/java/com/intel/jndn/utils/RepositoryServer.java
rename to src/main/java/com/intel/jndn/utils/server/RepositoryServer.java
index 88219d8..09ebc23 100644
--- a/src/main/java/com/intel/jndn/utils/RepositoryServer.java
+++ b/src/main/java/com/intel/jndn/utils/server/RepositoryServer.java
@@ -11,9 +11,9 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils;
+package com.intel.jndn.utils.server;
-import com.intel.jndn.utils.server.Server;
+import com.intel.jndn.utils.Server;
import java.io.IOException;
import net.named_data.jndn.Data;
diff --git a/src/main/java/com/intel/jndn/utils/SegmentedServer.java b/src/main/java/com/intel/jndn/utils/server/impl/SegmentedServer.java
similarity index 89%
rename from src/main/java/com/intel/jndn/utils/SegmentedServer.java
rename to src/main/java/com/intel/jndn/utils/server/impl/SegmentedServer.java
index 875ca87..6be4280 100644
--- a/src/main/java/com/intel/jndn/utils/SegmentedServer.java
+++ b/src/main/java/com/intel/jndn/utils/server/impl/SegmentedServer.java
@@ -11,12 +11,13 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils;
+package com.intel.jndn.utils.server.impl;
-import com.intel.jndn.utils.repository.ForLoopRepository;
-import com.intel.jndn.utils.repository.Repository;
-import com.intel.jndn.utils.server.SegmentedServerHelper;
-import com.intel.jndn.utils.server.ServerBaseImpl;
+import com.intel.jndn.utils.server.RepositoryServer;
+import com.intel.jndn.utils.repository.impl.ForLoopRepository;
+import com.intel.jndn.utils.Repository;
+import com.intel.jndn.utils.server.impl.SegmentedServerHelper;
+import com.intel.jndn.utils.server.impl.ServerBaseImpl;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
diff --git a/src/main/java/com/intel/jndn/utils/server/SegmentedServerHelper.java b/src/main/java/com/intel/jndn/utils/server/impl/SegmentedServerHelper.java
similarity index 98%
rename from src/main/java/com/intel/jndn/utils/server/SegmentedServerHelper.java
rename to src/main/java/com/intel/jndn/utils/server/impl/SegmentedServerHelper.java
index 2e1f92c..78e3c27 100644
--- a/src/main/java/com/intel/jndn/utils/server/SegmentedServerHelper.java
+++ b/src/main/java/com/intel/jndn/utils/server/impl/SegmentedServerHelper.java
@@ -11,7 +11,7 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.server;
+package com.intel.jndn.utils.server.impl;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
diff --git a/src/main/java/com/intel/jndn/utils/server/ServerBaseImpl.java b/src/main/java/com/intel/jndn/utils/server/impl/ServerBaseImpl.java
similarity index 90%
rename from src/main/java/com/intel/jndn/utils/server/ServerBaseImpl.java
rename to src/main/java/com/intel/jndn/utils/server/impl/ServerBaseImpl.java
index b69b455..6110005 100644
--- a/src/main/java/com/intel/jndn/utils/server/ServerBaseImpl.java
+++ b/src/main/java/com/intel/jndn/utils/server/impl/ServerBaseImpl.java
@@ -11,8 +11,10 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils.server;
+package com.intel.jndn.utils.server.impl;
+import com.intel.jndn.utils.ProcessingStage;
+import com.intel.jndn.utils.Server;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
@@ -36,7 +38,7 @@
private static final Logger logger = Logger.getLogger(ServerBaseImpl.class.getName());
private final Face face;
private final Name prefix;
- private final List<PipelineStage> pipeline = new ArrayList<>();
+ private final List<ProcessingStage> pipeline = new ArrayList<>();
private long registeredPrefixId = UNREGISTERED;
/**
@@ -89,7 +91,7 @@
logger.log(Level.SEVERE, "Failed to register prefix: " + prefix.toUri());
}
}, new ForwardingFlags());
- logger.log(Level.FINER, "Registered a new prefix: " + prefix.toUri());
+ logger.log(Level.FINER, "Registered a new prefix: " + prefix.toUri());
} catch (net.named_data.jndn.security.SecurityException e) {
throw new IOException("Failed to communicate to face due to security error", e);
}
@@ -99,13 +101,13 @@
* {@inheritDoc}
*/
@Override
- public void addPipelineStage(PipelineStage<Data, Data> pipelineStage) {
+ public void addPostProcessingStage(ProcessingStage<Data, Data> pipelineStage) {
pipeline.add(pipelineStage);
}
/**
* Process the {@link Data} before sending it; this runs the packet through
- * each registered {@link PipelineStage} in order.
+ * each registered {@link ProcessingStage} in order.
*
* @param data the {@link Data} to process
* @return a processed {@link Data} packet; no guarantee as to whether it is
@@ -113,7 +115,7 @@
* @throws Exception if a pipeline stage fails
*/
public Data processPipeline(Data data) throws Exception {
- for (PipelineStage<Data, Data> stage : pipeline) {
+ for (ProcessingStage<Data, Data> stage : pipeline) {
data = stage.process(data);
}
return data;
diff --git a/src/main/java/com/intel/jndn/utils/SimpleServer.java b/src/main/java/com/intel/jndn/utils/server/impl/SimpleServer.java
similarity index 90%
rename from src/main/java/com/intel/jndn/utils/SimpleServer.java
rename to src/main/java/com/intel/jndn/utils/server/impl/SimpleServer.java
index b983fa3..cee7d7f 100644
--- a/src/main/java/com/intel/jndn/utils/SimpleServer.java
+++ b/src/main/java/com/intel/jndn/utils/server/impl/SimpleServer.java
@@ -11,13 +11,15 @@
* FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for
* more details.
*/
-package com.intel.jndn.utils;
+package com.intel.jndn.utils.server.impl;
+import com.intel.jndn.utils.server.DynamicServer;
import com.intel.jndn.utils.server.RespondWithData;
import com.intel.jndn.utils.server.RespondWithBlob;
-import com.intel.jndn.utils.server.ServerBaseImpl;
+import com.intel.jndn.utils.server.RespondWithBlob;
+import com.intel.jndn.utils.server.RespondWithData;
+import com.intel.jndn.utils.server.impl.ServerBaseImpl;
import java.io.IOException;
-import java.nio.ByteBuffer;
import java.util.logging.Level;
import java.util.logging.Logger;
import net.named_data.jndn.Data;
@@ -25,7 +27,7 @@
import net.named_data.jndn.Interest;
import net.named_data.jndn.InterestFilter;
import net.named_data.jndn.Name;
-import net.named_data.jndn.transport.Transport;
+import net.named_data.jndn.OnInterest;
import net.named_data.jndn.util.Blob;
/**