Initial pubsub; waiting on MockFace to fully test
diff --git a/nb-configuration.xml b/nb-configuration.xml
index ae9d239..39376a3 100644
--- a/nb-configuration.xml
+++ b/nb-configuration.xml
@@ -1,18 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<project-shared-configuration>
- <!--
-This file contains additional configuration written by modules in the NetBeans IDE.
-The configuration is intended to be shared among all the users of project and
-therefore it is assumed to be part of version control checkout.
-Without this configuration present, some functionality in the IDE may be limited or fail altogether.
--->
- <properties xmlns="http://www.netbeans.org/ns/maven-properties-data/1">
- <!--
-Properties that influence various parts of the IDE, especially code formatting and the like.
-You can copy and paste the single properties, into the pom.xml file and the IDE will pick them up.
-That way multiple projects can share the same settings (useful for formatting rules for example).
-Any value defined here will override the pom.xml file value but is only applicable to the current project.
--->
- <netbeans.hint.license>intel-license</netbeans.hint.license>
- </properties>
+ <!--
+ This file contains additional configuration written by modules in the NetBeans IDE.
+ The configuration is intended to be shared among all the users of project and
+ therefore it is assumed to be part of version control checkout.
+ Without this configuration present, some functionality in the IDE may be limited or fail altogether.
+ -->
+ <properties xmlns="http://www.netbeans.org/ns/maven-properties-data/1">
+ <!--
+ Properties that influence various parts of the IDE, especially code formatting and the like.
+ You can copy and paste the single properties, into the pom.xml file and the IDE will pick them up.
+ That way multiple projects can share the same settings (useful for formatting rules for example).
+ Any value defined here will override the pom.xml file value but is only applicable to the current project.
+ -->
+ <netbeans.hint.license>intel-license</netbeans.hint.license>
+ <org-netbeans-modules-editor-indent.CodeStyle.usedProfile>project</org-netbeans-modules-editor-indent.CodeStyle.usedProfile>
+ <org-netbeans-modules-editor-indent.CodeStyle.project.spaces-per-tab>4</org-netbeans-modules-editor-indent.CodeStyle.project.spaces-per-tab>
+ <org-netbeans-modules-editor-indent.CodeStyle.project.tab-size>4</org-netbeans-modules-editor-indent.CodeStyle.project.tab-size>
+ <org-netbeans-modules-editor-indent.CodeStyle.project.indent-shift-width>4</org-netbeans-modules-editor-indent.CodeStyle.project.indent-shift-width>
+ <org-netbeans-modules-editor-indent.CodeStyle.project.expand-tabs>false</org-netbeans-modules-editor-indent.CodeStyle.project.expand-tabs>
+ <org-netbeans-modules-editor-indent.CodeStyle.project.text-limit-width>80</org-netbeans-modules-editor-indent.CodeStyle.project.text-limit-width>
+ <org-netbeans-modules-editor-indent.CodeStyle.project.text-line-wrap>none</org-netbeans-modules-editor-indent.CodeStyle.project.text-line-wrap>
+ <org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.spaces-per-tab>2</org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.spaces-per-tab>
+ <org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.tab-size>4</org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.tab-size>
+ <org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.indent-shift-width>2</org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.indent-shift-width>
+ <org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.expand-tabs>true</org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.expand-tabs>
+ <org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.text-limit-width>80</org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.text-limit-width>
+ <org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.text-line-wrap>none</org-netbeans-modules-editor-indent.text.x-java.CodeStyle.project.text-line-wrap>
+ <org-netbeans-modules-editor-indent.text.x-fortran.CodeStyle.project.text-limit-width>132</org-netbeans-modules-editor-indent.text.x-fortran.CodeStyle.project.text-limit-width>
+ </properties>
</project-shared-configuration>
diff --git a/src/main/java/com/intel/jndn/utils/AsyncResult.java b/src/main/java/com/intel/jndn/utils/AsyncResult.java
deleted file mode 100644
index 97bc1bd..0000000
--- a/src/main/java/com/intel/jndn/utils/AsyncResult.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * File name: AsyncRequest.java
- *
- * Purpose: Helper class for tracking asynchronous Interest requests
- *
- * © Copyright Intel Corporation. All rights reserved.
- * Intel Corporation, 2200 Mission College Boulevard,
- * Santa Clara, CA 95052-8119, USA
- */
-package com.intel.jndn.utils;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Observable;
-import net.named_data.jndn.Data;
-
-public class AsyncResult extends Observable {
-
- public int responses = 0;
- public boolean done = false;
- public boolean success;
- public List<Data> packets = new ArrayList<>();
-
- /**
- * Call this when the request has changed
- */
- public void changed() {
- this.setChanged();
- this.notifyObservers();
- this.clearChanged();
- }
-}
diff --git a/src/main/java/com/intel/jndn/utils/Client.java b/src/main/java/com/intel/jndn/utils/Client.java
index 62a6a36..269fe5c 100644
--- a/src/main/java/com/intel/jndn/utils/Client.java
+++ b/src/main/java/com/intel/jndn/utils/Client.java
@@ -31,297 +31,300 @@
*/
public class Client {
- public static final long DEFAULT_SLEEP_TIME = 20;
- public static final long DEFAULT_TIMEOUT = 2000;
- private static final Logger logger = LogManager.getLogger();
+ public static final long DEFAULT_SLEEP_TIME = 20;
+ public static final long DEFAULT_TIMEOUT = 2000;
+ private static final Logger logger = LogManager.getLogger();
- /**
- * 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
- */
- public Data getSync(Face face, Interest interest) {
- // setup event
- long startTime = System.currentTimeMillis();
- final ClientObservableEvent event = new ClientObservableEvent();
+ /**
+ * 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
+ */
+ public Data getSync(Face face, Interest interest) {
+ // setup event
+ long startTime = System.currentTimeMillis();
+ final ClientObservableEvent event = new ClientObservableEvent(); // this event is used without observer/observables for speed; just serves as a final reference into the callbacks
- // send interest
- try {
- face.expressInterest(interest, new OnData() {
- @Override
- public void onData(Interest interest, Data data) {
- event.setTimestamp(System.currentTimeMillis());
- event.setSuccess(true);
- event.setPacket(data);
- }
- }, new OnTimeout() {
- @Override
- public void onTimeout(Interest interest) {
- event.setTimestamp(System.currentTimeMillis());
- event.setSuccess(false);
- event.setPacket(new Object());
- }
- });
- } catch (IOException e) {
- logger.warn("IO failure while sending interest.", e);
- return null;
- }
+ // send interest
+ try {
+ face.expressInterest(interest, new OnData() {
+ @Override
+ public void onData(Interest interest, Data data) {
+ event.setTimestamp(System.currentTimeMillis());
+ event.setSuccess(true);
+ event.setPacket(data);
+ }
+ }, new OnTimeout() {
+ @Override
+ public void onTimeout(Interest interest) {
+ event.setTimestamp(System.currentTimeMillis());
+ event.setSuccess(false);
+ event.setPacket(new Object());
+ }
+ });
+ } catch (IOException e) {
+ logger.warn("IO failure while sending interest.", e);
+ return null;
+ }
- // process events until a response is received or timeout
- while (event.getPacket() == null) {
- try {
- synchronized (face) {
- face.processEvents();
- }
- } catch (IOException | EncodingException e) {
- logger.warn("Failed to process events.", e);
- return null;
- }
- sleep();
- }
+ // process events until a response is received or timeout
+ while (event.getPacket() == null) {
+ try {
+ synchronized (face) {
+ face.processEvents();
+ }
+ } catch (IOException | EncodingException e) {
+ logger.warn("Failed to process events.", e);
+ return null;
+ }
+ sleep();
+ }
- // return
- logger.debug("Request time (ms): " + (event.getTimestamp() - startTime));
- return (event.isSuccess()) ? (Data) event.getPacket() : null;
- }
+ // return
+ logger.debug("Request time (ms): " + (event.getTimestamp() - startTime));
+ return (event.isSuccess()) ? (Data) event.getPacket() : 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
- */
- public Data getSync(Face face, Name name) {
- return getSync(face, getDefaultInterest(name));
- }
+ /**
+ * 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 Data getSync(Face face, Name name) {
+ return getSync(face, getDefaultInterest(name));
+ }
- /**
- * Asynchronously retrieve the Data for a given interest; use the returned
- * ClientObserver to handle the Data when it arrives. E.g.: Client.get(face,
- * interest).then((data) -> doSomething(data));
- *
- * @param face
- * @param interest
- * @return
- */
- public ClientObserver get(final Face face, final Interest interest) {
- // setup observer
- final ClientObserver observer = new ClientObserver();
- final ClientObservable eventHandler = new ClientObservable();
- eventHandler.addObserver(observer);
+ /**
+ * Asynchronously retrieve the Data for a given interest; use the returned
+ * ClientObserver to handle the Data when it arrives. E.g.: Client.get(face,
+ * interest).then((data) -> doSomething(data));
+ *
+ * @param face
+ * @param interest
+ * @return
+ */
+ public ClientObserver get(final Face face, final Interest interest) {
+ // setup observer
+ final ClientObserver observer = new ClientObserver();
+ final ClientObservable eventHandler = new ClientObservable();
+ eventHandler.addObserver(observer);
- // setup background thread
- Thread backgroundThread = new Thread(new Runnable() {
- @Override
- public void run() {
- // send interest
- try {
- face.expressInterest(interest, eventHandler, eventHandler);
- } catch (IOException e) {
- logger.warn("IO failure while sending interest.", e);
- observer.update(eventHandler, new ClientObservableEvent());
- }
+ // setup background thread
+ Thread backgroundThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ // send interest
+ try {
+ face.expressInterest(interest, eventHandler, eventHandler);
+ } catch (IOException e) {
+ logger.warn("IO failure while sending interest.", e);
+ eventHandler.notify(e);
+ }
- // process events until a response is received or timeout
- while (observer.responses() == 0) {
- try {
- synchronized (face) {
- face.processEvents();
- }
- } catch (IOException | EncodingException e) {
- logger.warn("Failed to process events.", e);
- observer.update(eventHandler, new ClientObservableEvent());
- }
- sleep();
- }
- }
- });
- backgroundThread.setName(String.format("Client.get(%s)", interest.getName().toUri()));
- backgroundThread.setDaemon(true);
- backgroundThread.start();
+ // process events until a response is received or timeout
+ while (observer.responses() == 0 && observer.errors() == 0 && !observer.mustStop()) {
+ try {
+ synchronized (face) {
+ face.processEvents();
+ }
+ } catch (IOException | EncodingException e) {
+ logger.warn("Failed to process events.", e);
+ eventHandler.notify(e);
+ }
+ sleep();
+ }
+
+ // finished
+ logger.trace("Received response; stopping thread.");
+ }
+ });
+ backgroundThread.setName(String.format("Client.get(%s)", interest.getName().toUri()));
+ backgroundThread.setDaemon(true);
+ backgroundThread.start();
- // return
- return observer;
- }
+ // return
+ return observer;
+ }
- /**
- * Asynchronously retrieve the Data for a Name using default Interest
- * parameters; see get(Face, Interest) for examples.
- *
- * @param face
- * @param name
- * @return
- */
- public ClientObserver get(Face face, Name name) {
- return get(face, getDefaultInterest(name));
- }
+ /**
+ * Asynchronously retrieve the Data for a Name using default Interest
+ * parameters; see get(Face, Interest) for examples.
+ *
+ * @param face
+ * @param name
+ * @return
+ */
+ public ClientObserver get(Face face, Name name) {
+ return get(face, getDefaultInterest(name));
+ }
- /**
- * Synchronously serve a Data on the given face until one request accesses
- * the data; will return incoming Interest request. E.g.: Interest request =
- * Client.putSync(face, data);
- *
- * @param face
- * @param data
- * @return
- */
- public Interest putSync(Face face, final Data data) {
- // setup event
- long startTime = System.currentTimeMillis();
- final String dataName = data.getName().toUri();
- final ClientObservableEvent event = new ClientObservableEvent();
+ /**
+ * Synchronously serve a Data on the given face until one request accesses the
+ * data; will return incoming Interest request. E.g.: Interest request =
+ * Client.putSync(face, data);
+ *
+ * @param face
+ * @param data
+ * @return
+ */
+ public Interest putSync(Face face, final Data data) {
+ // setup event
+ long startTime = System.currentTimeMillis();
+ final String dataName = data.getName().toUri();
+ final ClientObservableEvent event = new ClientObservableEvent();
- // setup flags
- ForwardingFlags flags = new ForwardingFlags();
- flags.setCapture(true);
+ // setup flags
+ ForwardingFlags flags = new ForwardingFlags();
+ flags.setCapture(true);
- // register the data name on the face
- try {
- face.registerPrefix(data.getName(), new OnInterest() {
- @Override
- public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
- try {
- transport.send(data.wireEncode().buf());
- logger.debug("Sent data: " + dataName);
- event.fromPacket(interest);
- } catch (IOException e) {
- logger.error("Failed to send data for: " + dataName);
- event.fromPacket(e);
- }
- }
- }, new OnRegisterFailed() {
- @Override
- public void onRegisterFailed(Name prefix) {
- event.fromPacket(new Exception("Failed to register name: " + dataName));
- }
- }, flags);
- logger.info("Registered data: " + dataName);
- } catch (IOException e) {
- logger.error("Could not connect to face to register prefix: " + dataName, e);
- event.fromPacket(e);
- } catch (net.named_data.jndn.security.SecurityException e) {
- logger.error("Error registering prefix: " + dataName, e);
- event.fromPacket(e);
- }
-
- // process events until one response is sent or error
- while (event.getPacket() == null) {
- try {
- synchronized (face) {
- face.processEvents();
- }
- } catch (IOException | EncodingException e) {
- logger.warn("Failed to process events.", e);
- event.fromPacket(e);
- }
- sleep();
- }
+ // register the data name on the face
+ try {
+ face.registerPrefix(data.getName(), new OnInterest() {
+ @Override
+ public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
+ try {
+ transport.send(data.wireEncode().buf());
+ logger.debug("Sent data: " + dataName);
+ event.fromPacket(interest);
+ } catch (IOException e) {
+ logger.error("Failed to send data for: " + dataName);
+ event.fromPacket(e);
+ }
+ }
+ }, new OnRegisterFailed() {
+ @Override
+ public void onRegisterFailed(Name prefix) {
+ event.fromPacket(new Exception("Failed to register name: " + dataName));
+ }
+ }, flags);
+ logger.info("Registered data: " + dataName);
+ } catch (IOException e) {
+ logger.error("Could not connect to face to register prefix: " + dataName, e);
+ event.fromPacket(e);
+ } catch (net.named_data.jndn.security.SecurityException e) {
+ logger.error("Error registering prefix: " + dataName, e);
+ event.fromPacket(e);
+ }
- // return
- logger.debug("Request time (ms): " + (event.getTimestamp() - startTime));
- return (event.isSuccess()) ? (Interest) event.getPacket() : null;
- }
+ // process events until one response is sent or error
+ while (event.getPacket() == null) {
+ try {
+ synchronized (face) {
+ face.processEvents();
+ }
+ } catch (IOException | EncodingException e) {
+ logger.warn("Failed to process events.", e);
+ event.fromPacket(e);
+ }
+ sleep();
+ }
- /**
- * Asynchronously serve a Data on the given face until an observer stops it.
- * E.g.: ClientObserver observer = Client.put(face, data); // when finished
- * serving the data, stop the background thread observer.stop();
- *
- * @param face
- * @param data
- * @return
- */
- public ClientObserver put(final Face face, final Data data) {
- // setup observer
- final ClientObserver observer = new ClientObserver();
- final ClientObservable eventHandler = new ClientObservable();
- eventHandler.addObserver(observer);
+ // return
+ logger.debug("Request time (ms): " + (event.getTimestamp() - startTime));
+ return (event.isSuccess()) ? (Interest) event.getPacket() : null;
+ }
- // setup handlers
- final OnInterest interestHandler = new OnInterest() {
- @Override
- public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
- try {
- transport.send(data.wireEncode().buf());
- } catch (IOException e) {
- logger.error("Failed to send data for: " + data.getName().toUri());
- eventHandler.onError(e);
- }
- }
- };
- final OnRegisterFailed failureHandler = new OnRegisterFailed() {
- @Override
- public void onRegisterFailed(Name prefix) {
- logger.error("Failed to register name to put: " + data.getName().toUri());
- eventHandler.onError(new Exception("Failed to register name to put: " + data.getName().toUri()));
- }
- };
- final ForwardingFlags flags = new ForwardingFlags();
- flags.setCapture(true);
+ /**
+ * Asynchronously serve a Data on the given face until an observer stops it.
+ * E.g.: ClientObserver observer = Client.put(face, data); // when finished
+ * serving the data, stop the background thread observer.stop();
+ *
+ * @param face
+ * @param data
+ * @return
+ */
+ public ClientObserver put(final Face face, final Data data) {
+ // setup observer
+ final ClientObserver observer = new ClientObserver();
+ final ClientObservable eventHandler = new ClientObservable();
+ eventHandler.addObserver(observer);
- // setup background thread
- Thread backgroundThread = new Thread(new Runnable() {
- @Override
- public void run() {
- // register name on the face
- try {
- face.registerPrefix(data.getName(), interestHandler, failureHandler, flags);
- logger.info("Registered data : " + data.getName().toUri());
- } catch (IOException e) {
- logger.error("Could not connect to face to register prefix: " + data.getName().toUri(), e);
- eventHandler.onError(e);
- } catch (net.named_data.jndn.security.SecurityException e) {
- logger.error("Error registering prefix: " + data.getName().toUri(), e);
- eventHandler.onError(e);
- }
+ // setup handlers
+ final OnInterest interestHandler = new OnInterest() {
+ @Override
+ public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
+ try {
+ transport.send(data.wireEncode().buf());
+ } catch (IOException e) {
+ logger.error("Failed to send data for: " + data.getName().toUri());
+ eventHandler.onError(e);
+ }
+ }
+ };
+ final OnRegisterFailed failureHandler = new OnRegisterFailed() {
+ @Override
+ public void onRegisterFailed(Name prefix) {
+ logger.error("Failed to register name to put: " + data.getName().toUri());
+ eventHandler.onError(new Exception("Failed to register name to put: " + data.getName().toUri()));
+ }
+ };
+ final ForwardingFlags flags = new ForwardingFlags();
+ flags.setCapture(true);
- // process events until a request is received
- while (observer.requests() == 0 && observer.errors() == 0 && !observer.mustStop()) {
- try {
- synchronized (face) {
- face.processEvents();
- }
- } catch (IOException | EncodingException e) {
- logger.warn("Failed to process events.", e);
- observer.update(eventHandler, new ClientObservableEvent());
- }
- sleep();
- }
- }
- });
- backgroundThread.setName(String.format("Client.put(%s)", data.getName().toUri()));
- backgroundThread.setDaemon(true);
- backgroundThread.start();
+ // setup background thread
+ Thread backgroundThread = new Thread(new Runnable() {
+ @Override
+ public void run() {
+ // register name on the face
+ try {
+ face.registerPrefix(data.getName(), interestHandler, failureHandler, flags);
+ logger.info("Registered data : " + data.getName().toUri());
+ } catch (IOException e) {
+ logger.error("Could not connect to face to register prefix: " + data.getName().toUri(), e);
+ eventHandler.onError(e);
+ } catch (net.named_data.jndn.security.SecurityException e) {
+ logger.error("Error registering prefix: " + data.getName().toUri(), e);
+ eventHandler.onError(e);
+ }
- return observer;
- }
+ // process events until a request is received
+ while (observer.requests() == 0 && observer.errors() == 0 && !observer.mustStop()) {
+ try {
+ synchronized (face) {
+ face.processEvents();
+ }
+ } catch (IOException | EncodingException e) {
+ logger.warn("Failed to process events.", e);
+ observer.update(eventHandler, new ClientObservableEvent());
+ }
+ sleep();
+ }
+ }
+ });
+ backgroundThread.setName(String.format("Client.put(%s)", data.getName().toUri()));
+ backgroundThread.setDaemon(true);
+ backgroundThread.start();
- /**
- * Put the current thread to sleep to allow time for IO
- */
- protected void sleep() {
- try {
- Thread.currentThread().sleep(DEFAULT_SLEEP_TIME);
- } catch (InterruptedException e) {
- logger.error("Event loop interrupted.", e);
- }
- }
+ return observer;
+ }
- /**
- * Create a default interest for a given Name using some common settings: -
- * lifetime: 2 seconds
- *
- * @param name
- * @return
- */
- public Interest getDefaultInterest(Name name) {
- Interest interest = new Interest(name, DEFAULT_TIMEOUT);
- return interest;
- }
+ /**
+ * Put the current thread to sleep to allow time for IO
+ */
+ protected void sleep() {
+ try {
+ Thread.currentThread().sleep(DEFAULT_SLEEP_TIME);
+ } catch (InterruptedException e) {
+ logger.error("Event loop interrupted.", e);
+ }
+ }
+
+ /**
+ * Create a default interest for a given Name using some common settings: -
+ * lifetime: 2 seconds
+ *
+ * @param name
+ * @return
+ */
+ public Interest getDefaultInterest(Name name) {
+ Interest interest = new Interest(name, DEFAULT_TIMEOUT);
+ return interest;
+ }
}
diff --git a/src/main/java/com/intel/jndn/utils/ClientObservable.java b/src/main/java/com/intel/jndn/utils/ClientObservable.java
index add1c29..ce6beda 100644
--- a/src/main/java/com/intel/jndn/utils/ClientObservable.java
+++ b/src/main/java/com/intel/jndn/utils/ClientObservable.java
@@ -26,26 +26,105 @@
*/
public class ClientObservable extends Observable implements OnData, OnTimeout, OnInterest {
- protected List<ClientObservableEvent> events = new ArrayList<>();
- protected List<Interest> incomingInterestPackets = new ArrayList<>();
- protected List<Data> incomingDataPackets;
+ protected List<ClientObservableEvent> events = new ArrayList<>();
+ protected List<Interest> incomingInterestPackets = new ArrayList<>();
+ protected List<Data> incomingDataPackets;
- @Override
- public void onData(Interest interest, Data data) {
- notifyObservers(new ClientObservableEvent(data));
- }
-
- public void onError(Exception e){
- notifyObservers(new ClientObservableEvent(e));
- }
+ /**
+ * Generic notification
+ *
+ * @param <T>
+ * @param packet
+ */
+ public <T> void notify(T packet) {
+ setChanged();
+ notifyObservers(new ClientObservableEvent(packet));
+ }
- @Override
- public void onTimeout(Interest interest) {
- notifyObservers(new ClientObservableEvent());
- }
-
- @Override
- public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
- notifyObservers(new ClientObservableEvent(interest));
- }
+ /**
+ * Handle data packets
+ *
+ * @param interest
+ * @param data
+ */
+ @Override
+ public void onData(Interest interest, Data data) {
+ notify(data);
+ }
+
+ /**
+ * Handle exceptions
+ *
+ * @param e
+ */
+ public void onError(Exception e) {
+ notify(e);
+ }
+
+ /**
+ * Handle timeouts
+ *
+ * @param interest
+ */
+ @Override
+ public void onTimeout(Interest interest) {
+ notify(interest);
+ }
+
+ /**
+ * Handle incoming interests
+ *
+ * @param prefix
+ * @param interest
+ * @param transport
+ * @param registeredPrefixId
+ */
+ @Override
+ public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
+ notify(interest); // TODO wrap
+ }
+
+ /**
+ * Helper to reference both outgoing interest and incoming data packets
+ */
+ class InterestDataPacket {
+
+ private Interest interest;
+ private Data data;
+
+ public InterestDataPacket(Interest interest, Data data) {
+ this.interest = interest;
+ this.data = data;
+ }
+
+ public Data getData() {
+ return data;
+ }
+
+ public Interest getInterest() {
+ return interest;
+ }
+ }
+
+ /**
+ * Helper to reference both incoming interest and the transport to send data on
+ */
+ class InterestTransportPacket{
+ private Interest interest;
+ private Transport transport;
+
+ public InterestTransportPacket(Interest interest, Transport transport) {
+ this.interest = interest;
+ this.transport = transport;
+ }
+
+ public Interest getInterest() {
+ return interest;
+ }
+
+ public Transport getTransport() {
+ return transport;
+ }
+
+ }
}
diff --git a/src/main/java/com/intel/jndn/utils/ClientObservableEvent.java b/src/main/java/com/intel/jndn/utils/ClientObservableEvent.java
index 7102f93..af2a614 100644
--- a/src/main/java/com/intel/jndn/utils/ClientObservableEvent.java
+++ b/src/main/java/com/intel/jndn/utils/ClientObservableEvent.java
@@ -15,47 +15,47 @@
*/
public class ClientObservableEvent<T> {
- boolean success;
- long timestamp;
- T packet;
+ boolean success;
+ long timestamp;
+ T packet;
- public ClientObservableEvent(T packet) {
- fromPacket(packet);
- }
+ public ClientObservableEvent(T packet) {
+ fromPacket(packet);
+ }
- public ClientObservableEvent() {
- timestamp = System.currentTimeMillis();
- success = false;
- }
-
- public final void fromPacket(T packet_){
- timestamp = System.currentTimeMillis();
- success = !Exception.class.isInstance(packet_);
- packet = packet_;
- }
+ public ClientObservableEvent() {
+ timestamp = System.currentTimeMillis();
+ success = false;
+ }
- public boolean isSuccess() {
- return success;
- }
+ public final void fromPacket(T packet_) {
+ timestamp = System.currentTimeMillis();
+ success = !Exception.class.isInstance(packet_);
+ packet = packet_;
+ }
- public void setSuccess(boolean success) {
- this.success = success;
- }
+ public boolean isSuccess() {
+ return success;
+ }
- public long getTimestamp() {
- return timestamp;
- }
+ public void setSuccess(boolean success) {
+ this.success = success;
+ }
- public void setTimestamp(long timestamp) {
- this.timestamp = timestamp;
- }
+ public long getTimestamp() {
+ return timestamp;
+ }
- public T getPacket() {
- return packet;
- }
+ public void setTimestamp(long timestamp) {
+ this.timestamp = timestamp;
+ }
- public void setPacket(T packet) {
- this.packet = packet;
- }
+ public T getPacket() {
+ return packet;
+ }
+
+ public void setPacket(T packet) {
+ this.packet = packet;
+ }
}
diff --git a/src/main/java/com/intel/jndn/utils/ClientObserver.java b/src/main/java/com/intel/jndn/utils/ClientObserver.java
index ed78476..0ed4e11 100644
--- a/src/main/java/com/intel/jndn/utils/ClientObserver.java
+++ b/src/main/java/com/intel/jndn/utils/ClientObserver.java
@@ -23,81 +23,137 @@
*/
public class ClientObserver implements Observer {
- protected List<ClientObservableEvent> events = new ArrayList<>();
- protected long timestamp;
- protected OnData then;
- protected boolean stopThread;
-
- public ClientObserver(){
- timestamp = System.currentTimeMillis();
- }
+ protected List<ClientObservableEvent> events = new ArrayList<>();
+ protected long timestamp;
+ protected OnData then;
+ protected boolean stopThread;
- @Override
- public void update(Observable o, Object arg) {
- ClientObservableEvent event = (ClientObservableEvent) arg;
- events.add(event);
- if(Data.class.isInstance(event.packet)){
- then.onData(null, (Data) event.packet);
- }
- }
-
- public ClientObserver then(OnData handler){
- then = handler;
- return this;
- }
+ /**
+ * Constructor
+ */
+ public ClientObserver() {
+ timestamp = System.currentTimeMillis();
+ }
- public int count(Class type) {
- int count = 0;
- for (ClientObservableEvent event : events) {
- if (type.isInstance(event.packet)) {
- count++;
- }
- }
- return count;
- }
+ /**
+ * Receive notifications from observables
+ *
+ * @param o
+ * @param arg
+ */
+ @Override
+ public void update(Observable o, Object arg) {
+ ClientObservableEvent event = (ClientObservableEvent) arg;
+ events.add(event);
+ // call onData callbacks
+ if (Data.class.isInstance(event.packet) && then != null) {
+ then.onData(null, (Data) event.packet);
+ }
+ }
- public int requests() {
- return count(Interest.class);
- }
+ /**
+ * Register a handler for data packets
+ *
+ * @param handler
+ * @return
+ */
+ public ClientObserver then(OnData handler) {
+ then = handler;
+ return this;
+ }
- public int responses() {
- return count(Data.class);
- }
-
- public int errors(){
- return count(Exception.class);
- }
-
- /**
- * Get time since observer start
- * @return
- */
- public long getTimeSinceStart(){
- if(getLast() != null){
- return getLast().getTimestamp() - timestamp;
- }
- return -1;
- }
+ /**
+ * Count the number of observed packets by type
+ *
+ * @param type
+ * @return
+ */
+ public int count(Class type) {
+ int count = 0;
+ for (ClientObservableEvent event : events) {
+ if (type.isInstance(event.packet)) {
+ count++;
+ }
+ }
+ return count;
+ }
- public ClientObservableEvent getFirst() {
- if (events.size() > 0) {
- return events.get(0);
- }
- return null;
- }
+ /**
+ * Count the number of interest packets observed (received or sent)
+ *
+ * @return
+ */
+ public int requests() {
+ return count(Interest.class);
+ }
- public ClientObservableEvent getLast() {
- if (events.size() > 0) {
- return events.get(events.size() - 1);
- }
- return null;
- }
-
- public void stop(){
- stopThread = true;
- }
-
- public boolean mustStop(){
- return stopThread;
- }
+ /**
+ * Count the number of Data packets observed
+ *
+ * @return
+ */
+ public int responses() {
+ return count(Data.class);
+ }
+
+ /**
+ * Count the number of errors
+ *
+ * @return
+ */
+ public int errors() {
+ return count(Exception.class);
+ }
+
+ /**
+ * Get time since observer start
+ *
+ * @return
+ */
+ public long getTimeSinceStart() {
+ if (getLast() != null) {
+ return getLast().getTimestamp() - timestamp;
+ }
+ return -1;
+ }
+
+ /**
+ * Retrieve first event
+ *
+ * @return event or null
+ */
+ public ClientObservableEvent getFirst() {
+ if (events.size() > 0) {
+ return events.get(0);
+ }
+ return null;
+ }
+
+ /**
+ * Retrieve last event
+ *
+ * @return event or null
+ */
+ public ClientObservableEvent getLast() {
+ if (events.size() > 0) {
+ return events.get(events.size() - 1);
+ }
+ return null;
+ }
+
+ /**
+ * Stop the current Client thread
+ */
+ public void stop() {
+ stopThread = true;
+ }
+
+ /**
+ * Check the current stop status
+ *
+ * @return
+ */
+ public boolean mustStop() {
+ return stopThread;
+ }
}
diff --git a/src/main/java/com/intel/jndn/utils/InternalFace.java b/src/main/java/com/intel/jndn/utils/InternalFace.java
new file mode 100644
index 0000000..980a457
--- /dev/null
+++ b/src/main/java/com/intel/jndn/utils/InternalFace.java
@@ -0,0 +1,19 @@
+/*
+ * File name: InternalFace.java
+ *
+ * Purpose: Handle data traffic within an application storing all packets in
+ * memory; useful for standardizing methods for getting/putting data
+ *
+ * © Copyright Intel Corporation. All rights reserved.
+ * Intel Corporation, 2200 Mission College Boulevard,
+ * Santa Clara, CA 95052-8119, USA
+ */
+package com.intel.jndn.utils;
+
+/**
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class InternalFace {
+
+}
diff --git a/src/main/java/com/intel/jndn/utils/Publisher.java b/src/main/java/com/intel/jndn/utils/Publisher.java
index 862f001..0ae79b1 100644
--- a/src/main/java/com/intel/jndn/utils/Publisher.java
+++ b/src/main/java/com/intel/jndn/utils/Publisher.java
@@ -41,175 +41,188 @@
*/
public class Publisher implements OnInterest, OnRegisterFailed {
- public static final long DEFAULT_TIMEOUT = 2000;
- private static final Logger logger = LogManager.getLogger();
- protected HashMap<Class, String> types = new HashMap<>();
- protected HashMap<Name, PublishedObject> published = new HashMap<>();
- protected Face face;
- protected Name channel;
- protected boolean registered = true;
- protected long publishLifetime = -1; // in milliseconds, -1 will persist as long as this instance lives
+ public static final long DEFAULT_TIMEOUT = 2000;
+ private static final Logger logger = LogManager.getLogger();
+ protected HashMap<Class, String> types = new HashMap<>();
+ protected HashMap<Name, PublishedObject> published = new HashMap<>();
+ protected Face face;
+ protected Name channel;
+ protected boolean registered = true;
+ protected long publishLifetime = -1; // in milliseconds, -1 will persist as long as this instance lives
- /**
- * Initialize a publisher on the given channel; published objects will
- * persist as long as this instance lives.
- *
- * @param face
- * @param channel
- */
- public Publisher(final Face face, final Name channel) {
- this.face = face;
- this.channel = channel;
- try {
- this.face.registerPrefix(this.channel, this, this, new ForwardingFlags());
- } catch (IOException e) {
- logger.error("Failed to send prefix registration for: " + this.channel.toUri(), e);
- } catch (SecurityException e) {
- logger.error("Failed to configure security correctly for registration: " + this.channel.toUri(), e);
- }
- }
+ /**
+ * Initialize a publisher on the given channel; published objects will persist
+ * as long as this instance lives.
+ *
+ * @param face
+ * @param channel
+ * @param registerOnForwarder set this to false to disable interaction with the forwarder (e.g. as in tests)
+ */
+ public Publisher(final Face face, final Name channel, boolean registerOnForwarder) {
+ this.face = face;
+ this.channel = channel;
+ if (registerOnForwarder) {
+ try {
+ this.face.registerPrefix(this.channel, this, this, new ForwardingFlags());
+ } catch (IOException e) {
+ logger.error("Failed to send prefix registration for: " + this.channel.toUri(), e);
+ } catch (SecurityException e) {
+ logger.error("Failed to configure security correctly for registration: " + this.channel.toUri(), e);
+ }
+ }
+ }
- /**
- * Initialize a publisher on the given channel; limit the lifetime of
- * published objects to save memory.
- *
- * @param face
- * @param channel
- * @param publishLifetime
- */
- public Publisher(final Face face, final Name channel, long publishLifetime) {
- this(face, channel);
- this.publishLifetime = publishLifetime;
- }
+ /**
+ * Initialize a publisher on the given channel; published objects will persist
+ * as long as this instance lives.
+ *
+ * @param face
+ * @param channel
+ */
+ public Publisher(final Face face, final Name channel) {
+ this(face, channel, true);
+ }
- /**
- * Add a type and its alias to this publisher
- *
- * @param typeAlias
- * @param type
- */
- public void addType(String typeAlias, Class type) {
- types.put(type, typeAlias);
- }
+ /**
+ * Initialize a publisher on the given channel; limit the lifetime of
+ * published objects to save memory.
+ *
+ * @param face
+ * @param channel
+ * @param publishLifetime
+ */
+ public Publisher(final Face face, final Name channel, long publishLifetime) {
+ this(face, channel);
+ this.publishLifetime = publishLifetime;
+ }
+ /**
+ * Add a type and its alias to this publisher
+ *
+ * @param typeAlias
+ * @param type
+ */
+ public void addType(String typeAlias, Class type) {
+ types.put(type, typeAlias);
+ }
- /**
- * Publish the given object by sending an alert to all subscribers
- *
- * @param publishedObject
- */
- public void publish(Object publishedObject) {
- // check if this publisher is registered on the face
- if (!registered) {
- logger.error("Publisher failed to register; cannot publish data to: " + channel.toUri());
- return;
- }
+ /**
+ * Publish the given object by sending an alert to all subscribers
+ *
+ * @param publishedObject
+ */
+ public void publish(Object publishedObject) {
+ // check if this publisher is registered on the face
+ if (!registered) {
+ logger.error("Publisher failed to register; cannot publish data to: " + channel.toUri());
+ return;
+ }
- // check known types
- if (!types.containsKey(publishedObject.getClass())) {
- logger.error("Attempted (and failed) to publish unknown type: " + publishedObject.getClass().getName());
- return;
- }
+ // check known types
+ if (!types.containsKey(publishedObject.getClass())) {
+ logger.error("Attempted (and failed) to publish unknown type: " + publishedObject.getClass().getName());
+ return;
+ }
- // check if old published objects need to be removed
- clearOldPublished();
+ // check if old published objects need to be removed
+ clearOldPublished();
- // track published objects
- Name publishedName = new Name(channel).append(types.get(publishedObject.getClass())).appendTimestamp(System.currentTimeMillis());
- published.put(publishedName, new PublishedObject(publishedObject));
+ // track published objects
+ Name publishedName = new Name(channel).append(types.get(publishedObject.getClass())).appendTimestamp(System.currentTimeMillis());
+ published.put(publishedName, new PublishedObject(publishedObject));
- // send out alert and don't wait for response
- Client client = new Client();
- client.get(face, getAlert(publishedName));
- logger.debug("Notified channel of new published object: " + publishedName.toUri());
- }
+ // send out alert and don't wait for response
+ Client client = new Client();
+ client.get(face, getAlert(publishedName));
+ logger.debug("Notified channel of new published object: " + publishedName.toUri());
+ }
- /**
- * Handle incoming requests for published objects; will check the published
- * list and respond with objects that exactly match the incoming interest.
- * TODO signing, encryption
- *
- * @param prefix
- * @param interest
- * @param transport
- * @param registeredPrefixId
- */
- @Override
- public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
- // check for old published objects
- clearOldPublished();
+ /**
+ * Handle incoming requests for published objects; will check the published
+ * list and respond with objects that exactly match the incoming interest.
+ * TODO signing, encryption
+ *
+ * @param prefix
+ * @param interest
+ * @param transport
+ * @param registeredPrefixId
+ */
+ @Override
+ public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
+ // check for old published objects
+ clearOldPublished();
- // publish object if available
- if (published.containsKey(interest.getName())) {
- try {
- Data data = new Data(interest.getName());
- data.setContent(new Blob("TODO parse object"));
- if (publishLifetime != -1) {
- data.getMetaInfo().setFreshnessPeriod(publishLifetime);
- }
- // data.setSignature(...);
- transport.send(data.wireEncode().signedBuf());
- logger.debug("Sent data: " + interest.getName().toUri());
- } catch (IOException e) {
- logger.error("Failed to send data: " + interest.getName().toUri());
- }
- }
- }
+ // publish object if available
+ if (published.containsKey(interest.getName())) {
+ try {
+ Data data = new Data(interest.getName());
+ data.setContent(new Blob("TODO parse object"));
+ if (publishLifetime != -1) {
+ data.getMetaInfo().setFreshnessPeriod(publishLifetime);
+ }
+ // data.setSignature(...);
+ transport.send(data.wireEncode().signedBuf());
+ logger.debug("Sent data: " + interest.getName().toUri());
+ } catch (IOException e) {
+ logger.error("Failed to send data: " + interest.getName().toUri());
+ }
+ }
+ }
- /**
- * Handle registration failure; this will stop the publisher from sending
- * notifications since the routes will not be setup to respond
- *
- * @param prefix
- */
- @Override
- public void onRegisterFailed(Name prefix) {
- logger.error("Failed to register publisher: " + channel.toUri());
- registered = false;
- }
+ /**
+ * Handle registration failure; this will stop the publisher from sending
+ * notifications since the routes will not be setup to respond
+ *
+ * @param prefix
+ */
+ @Override
+ public void onRegisterFailed(Name prefix) {
+ logger.error("Failed to register publisher: " + channel.toUri());
+ registered = false;
+ }
- /**
- * Remove any published objects that have outlived their lifetime; a
- * lifetime of -1 persists them forever.
- */
- public void clearOldPublished() {
- if (publishLifetime == -1) {
- return;
- }
- for (Entry<Name, PublishedObject> e : published.entrySet()) {
- if (System.currentTimeMillis() - e.getValue().publishedOn > publishLifetime) {
- published.remove(e.getKey());
- logger.debug("Removing old published object: " + e.getKey().toUri());
- }
- }
- }
-
- /**
- * Build an alert Interest to notify subscribers to retrieve data from this
- * publisher; the Interest will have the form
- * /this/published/channel/ALERT/[encoded channel to request]
- *
- * @param nameToPublish
- * @return
- */
- protected Interest getAlert(Name nameToPublish) {
- Name alertName = new Name(channel).append("ALERT").append(nameToPublish.wireEncode());
- Interest alert = new Interest(alertName);
- alert.setMustBeFresh(true); // interest must reach the application subscribers
- alert.setInterestLifetimeMilliseconds(DEFAULT_TIMEOUT);
- return alert;
- }
+ /**
+ * Remove any published objects that have outlived their lifetime; a lifetime
+ * of -1 persists them forever.
+ */
+ public void clearOldPublished() {
+ if (publishLifetime == -1) {
+ return;
+ }
+ for (Entry<Name, PublishedObject> e : published.entrySet()) {
+ if (System.currentTimeMillis() - e.getValue().publishedOn > publishLifetime) {
+ published.remove(e.getKey());
+ logger.debug("Removing old published object: " + e.getKey().toUri());
+ }
+ }
+ }
- /**
- * Helper class to track published objects and their timestamps
- */
- private class PublishedObject {
+ /**
+ * Build an alert Interest to notify subscribers to retrieve data from this
+ * publisher; the Interest will have the form
+ * /this/published/channel/ALERT/[encoded channel to request]
+ *
+ * @param nameToPublish
+ * @return
+ */
+ protected Interest getAlert(Name nameToPublish) {
+ Name alertName = new Name(channel).append("ALERT").append(nameToPublish.wireEncode());
+ Interest alert = new Interest(alertName);
+ alert.setMustBeFresh(true); // interest must reach the application subscribers
+ alert.setInterestLifetimeMilliseconds(DEFAULT_TIMEOUT);
+ return alert;
+ }
- Object publishedObject;
- long publishedOn;
+ /**
+ * Helper class to track published objects and their timestamps
+ */
+ private class PublishedObject {
- public PublishedObject(Object object) {
- publishedObject = object;
- }
- }
+ Object publishedObject;
+ long publishedOn;
+
+ public PublishedObject(Object object) {
+ publishedObject = object;
+ }
+ }
}
diff --git a/src/main/java/com/intel/jndn/utils/Subscriber.java b/src/main/java/com/intel/jndn/utils/Subscriber.java
index 23e8cf6..df6d375 100644
--- a/src/main/java/com/intel/jndn/utils/Subscriber.java
+++ b/src/main/java/com/intel/jndn/utils/Subscriber.java
@@ -40,179 +40,193 @@
*/
public class Subscriber implements OnInterest, OnRegisterFailed {
- public static final long DEFAULT_TIMEOUT = 2000;
- private static final Logger logger = LogManager.getLogger();
- protected HashMap<String, Class> types = new HashMap<>();
- protected HashMap<Class, OnPublish> handlers = new HashMap<>();
- protected Face face;
- protected Name channel;
+ public static final long DEFAULT_TIMEOUT = 2000;
+ private static final Logger logger = LogManager.getLogger();
+ protected HashMap<String, Class> types = new HashMap<>();
+ protected HashMap<Class, OnPublish> handlers = new HashMap<>();
+ protected Face face;
+ protected Name channel;
- /**
- * Initialize a publisher on the given channel; published objects will
- * persist as long as this instance lives.
- *
- * @param face
- * @param channel
- */
- public Subscriber(final Face face, final Name channel) {
- this.face = face;
- this.channel = channel;
- try {
- this.face.registerPrefix(this.channel, this, this, new ForwardingFlags());
- } catch (IOException e) {
- logger.error("Failed to send prefix registration for: " + this.channel.toUri(), e);
- } catch (net.named_data.jndn.security.SecurityException e) {
- logger.error("Failed to configure security correctly for registration: " + this.channel.toUri(), e);
- }
- }
+ /**
+ * Initialize a publisher on the given channel; published objects will persist
+ * as long as this instance lives.
+ *
+ * @param face
+ * @param channel
+ * @param registerOnForwarder set this to false to disable interaction with the forwarder (e.g. as in tests)
+ */
+ public Subscriber(final Face face, final Name channel, boolean registerOnForwarder) {
+ this.face = face;
+ this.channel = channel;
+ if (registerOnForwarder) {
+ try {
+ this.face.registerPrefix(this.channel, this, this, new ForwardingFlags());
+ } catch (IOException e) {
+ logger.error("Failed to send prefix registration for: " + this.channel.toUri(), e);
+ } catch (net.named_data.jndn.security.SecurityException e) {
+ logger.error("Failed to configure security correctly for registration: " + this.channel.toUri(), e);
+ }
+ }
+ }
- /**
- * Add a type and its alias to this publisher
- *
- * @param typeAlias
- * @param type
- */
- public void addType(String typeAlias, Class type) {
- types.put(typeAlias, type);
- }
-
-
- /**
- * Functional interface defining action to take upon receipt of a published
- * object
- *
- * @param <T>
- */
- public interface OnPublish<T>{
- public void onPublish(T publishedObject);
- }
+ /**
+ * Initialize a publisher on the given channel; published objects will persist
+ * as long as this instance lives.
+ *
+ * @param face
+ * @param channel
+ */
+ public Subscriber(final Face face, final Name channel) {
+ this(face, channel, true);
+ }
- /**
- * Register a handler for receipt of a certain published type
- *
- * @param type
- * @param handler
- */
- public void on(Class type, OnPublish handler) {
- handlers.put(type, handler);
- }
+ /**
+ * Add a type and its alias to this publisher
+ *
+ * @param typeAlias
+ * @param type
+ */
+ public void addType(String typeAlias, Class type) {
+ types.put(typeAlias, type);
+ }
- /**
- * Register a handler for receipt of a certain published type
- *
- * @param typeAlias
- * @param handler
- */
- public void on(String typeAlias, OnPublish handler) {
- if (types.containsKey(typeAlias)) {
- handlers.put(types.get(typeAlias), handler);
- } else {
- logger.warn("Unrecognized type (no handler registered): " + typeAlias);
- }
- }
+ /**
+ * Functional interface defining action to take upon receipt of a published
+ * object
+ *
+ * @param <T>
+ */
+ public interface OnPublish<T> {
- /**
- * Handle alert notifications for published objects; once the published
- * object is parsed, this will pass the object to the handle registered
- * in on().
- *
- * @param prefix
- * @param interest
- * @param transport
- * @param registeredPrefixId
- */
- @Override
- public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
- // check format
- if (!isAlert(interest)) {
- logger.warn("Incoming interest was not in ALERT format: " + interest.getName().toUri());
- return;
- }
+ public void onPublish(T publishedObject);
+ }
- // check signature
- // TODO
- // retrieve name
- Name publishedName = getAlertName(interest);
- if (publishedName == null) {
- return;
- }
+ /**
+ * Register a handler for receipt of a certain published type
+ *
+ * @param <T>
+ * @param type
+ * @param handler
+ */
+ public <T> void on(Class<T> type, OnPublish<T> handler) {
+ handlers.put(type, handler);
+ }
- // retrieve data
- Client client = new Client();
- Data data = client.getSync(face, publishedName);
- if (data == null) {
- logger.warn("Faled to retrieve published object: " + publishedName.toUri());
- return;
- }
+ /**
+ * Register a handler for receipt of a certain published type
+ *
+ * @param typeAlias
+ * @param handler
+ */
+ public void on(String typeAlias, OnPublish handler) {
+ if (types.containsKey(typeAlias)) {
+ handlers.put(types.get(typeAlias), handler);
+ } else {
+ logger.warn("Unrecognized type (no handler registered): " + typeAlias);
+ }
+ }
- // get handler
- String typeAlias = getPublishedType(data);
- if (!types.containsKey(typeAlias)) {
- logger.warn("Type not found: " + typeAlias);
- return;
- }
- Class type = types.get(typeAlias);
- if (!handlers.containsKey(type)) {
- logger.warn("No handler found for type: " + typeAlias);
- return;
- }
+ /**
+ * Handle alert notifications for published objects; once the published object
+ * is parsed, this will pass the object to the handle registered in on().
+ *
+ * @param prefix
+ * @param interest
+ * @param transport
+ * @param registeredPrefixId
+ */
+ @Override
+ public void onInterest(Name prefix, Interest interest, Transport transport, long registeredPrefixId) {
+ // check format
+ if (!isAlert(interest)) {
+ logger.warn("Incoming interest was not in ALERT format: " + interest.getName().toUri());
+ return;
+ }
- // build object
- Object publishedObject = null;
+ // check signature
+ // TODO
+ // retrieve name
+ Name publishedName = getAlertName(interest);
+ if (publishedName == null) {
+ return;
+ }
+
+ // retrieve data
+ Client client = new Client();
+ Data data = client.getSync(face, publishedName);
+ if (data == null) {
+ logger.warn("Faled to retrieve published object: " + publishedName.toUri());
+ return;
+ }
+
+ // get handler
+ String typeAlias = getPublishedType(data);
+ if (!types.containsKey(typeAlias)) {
+ logger.warn("Type not found: " + typeAlias);
+ return;
+ }
+ Class type = types.get(typeAlias);
+ if (!handlers.containsKey(type)) {
+ logger.warn("No handler found for type: " + typeAlias);
+ return;
+ }
+
+ // build object
+ Object publishedObject = null;
// TODO parse object
- // call
- handlers.get(type).onPublish(publishedObject);
- }
+ // call
+ handlers.get(type).onPublish(type.cast(publishedObject));
+ }
- /**
- * Handle registration failure;
- *
- * @param prefix
- */
- @Override
- public void onRegisterFailed(Name prefix) {
- logger.error("Failed to register: " + prefix.toUri());
- }
+ /**
+ * Handle registration failure;
+ *
+ * @param prefix
+ */
+ @Override
+ public void onRegisterFailed(Name prefix) {
+ logger.error("Failed to register: " + prefix.toUri());
+ }
- /**
- * Check if an incoming interest is an alert notification to retrieve data
- *
- * @param interest
- * @return
- */
- protected boolean isAlert(Interest interest) {
- Component alertComponent = interest.getName().get(-2);
- return alertComponent == null || !alertComponent.equals(new Component("ALERT"));
- }
+ /**
+ * Check if an incoming interest is an alert notification to retrieve data
+ *
+ * @param interest
+ * @return
+ */
+ protected boolean isAlert(Interest interest) {
+ Component alertComponent = interest.getName().get(-2);
+ return alertComponent == null || !alertComponent.equals(new Component("ALERT"));
+ }
- /**
- * Parse an Interest to retrieve the remote name of the published object to
- * then request; the Interest must have the form
- * /this/published/channel/ALERT/[encoded name to request]
- *
- * @param interest
- * @return
- */
- protected Name getAlertName(Interest interest) {
- try {
- Name publishedName = new Name();
- publishedName.wireDecode(interest.getName().get(-1).getValue());
- return publishedName;
- } catch (EncodingException e) {
- logger.error("Failed to parse remote published name", e);
- return null;
- }
- }
+ /**
+ * Parse an Interest to retrieve the remote name of the published object to
+ * then request; the Interest must have the form
+ * /this/published/channel/ALERT/[encoded name to request]
+ *
+ * @param interest
+ * @return
+ */
+ protected Name getAlertName(Interest interest) {
+ try {
+ Name publishedName = new Name();
+ publishedName.wireDecode(interest.getName().get(-1).getValue());
+ return publishedName;
+ } catch (EncodingException e) {
+ logger.error("Failed to parse remote published name", e);
+ return null;
+ }
+ }
- /**
- * Parse the published type from a Data packet; must be the second to last
- * component of the name
- *
- * @param data
- * @return
- */
- protected String getPublishedType(Data data) {
- return data.getName().get(-2).toEscapedString();
- }
+ /**
+ * Parse the published type from a Data packet; must be the second to last
+ * component of the name
+ *
+ * @param data
+ * @return
+ */
+ protected String getPublishedType(Data data) {
+ return data.getName().get(-2).toEscapedString();
+ }
}
diff --git a/src/main/resources/log4j2-test.xml b/src/main/resources/log4j2-test.xml
index 602b5ab..badfbef 100644
--- a/src/main/resources/log4j2-test.xml
+++ b/src/main/resources/log4j2-test.xml
@@ -1,13 +1,13 @@
<?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="info">
- <AppenderRef ref="Console"/>
- </Root>
- </Loggers>
+ <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
index 0dac33a..66435f2 100644
--- a/src/main/resources/log4j2.xml
+++ b/src/main/resources/log4j2.xml
@@ -1,13 +1,13 @@
<?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>
+ <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 0ac93b5..62b67d5 100644
--- a/src/test/java/com/intel/jndn/utils/ClientTest.java
+++ b/src/test/java/com/intel/jndn/utils/ClientTest.java
@@ -20,50 +20,68 @@
import net.named_data.jndn.Face;
import net.named_data.jndn.Name;
import net.named_data.jndn.util.Blob;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
/**
*
* @author Andrew Brown <andrew.brown@intel.com>
*/
public class ClientTest {
+
+
+ /**
+ * Setup logging
+ */
+ private static final Logger logger = LogManager.getLogger();
- @Test
- public void testGetSync() {
- // setup face
- MockTransport transport = new MockTransport();
- Face face = new Face(transport, null);
+ /**
+ * Test retrieving data synchronously
+ */
+ @Test
+ public void testGetSync() {
+ // setup face
+ MockTransport transport = new MockTransport();
+ Face face = new Face(transport, null);
- // setup return data
- Data response = new Data(new Name("/a/b/c"));
- response.setContent(new Blob("..."));
- transport.respondWith(response);
+ // setup return data
+ Data response = new Data(new Name("/a/b/c"));
+ response.setContent(new Blob("..."));
+ transport.respondWith(response);
- // retrieve data
- Client client = new Client();
- Data data = client.getSync(face, new Name("/a/b/c"));
- assertEquals(new Blob("...").buf(), data.getContent().buf());
- }
+ // retrieve data
+ logger.info("Client expressing interest synchronously: /a/b/c");
+ Client client = new Client();
+ Data data = client.getSync(face, new Name("/a/b/c"));
+ assertEquals(new Blob("...").buf(), data.getContent().buf());
+ }
- @Test
- public void testGetAsync() throws InterruptedException {
- // setup face
- MockTransport transport = new MockTransport();
- Face face = new Face(transport, null);
+ /**
+ * Test retrieving data asynchronously
+ *
+ * @throws InterruptedException
+ */
+ @Test
+ public void testGetAsync() throws InterruptedException {
+ // setup face
+ MockTransport transport = new MockTransport();
+ Face face = new Face(transport, null);
- // setup return data
- Data response = new Data(new Name("/a/b/c"));
- response.setContent(new Blob("..."));
- transport.respondWith(response);
+ // setup return data
+ Data response = new Data(new Name("/a/b/c"));
+ response.setContent(new Blob("..."));
+ transport.respondWith(response);
- // retrieve data
- Client client = new Client();
- ClientObserver observer = client.get(face, new Name("/a/b/c"));
-
- // wait
- while(observer.responses() == 0){
- Thread.sleep(10);
- }
- Data data = (Data) observer.getFirst().getPacket();
- assertEquals(new Blob("...").buf(), data.getContent().buf());
- }
+ // retrieve data
+ logger.info("Client expressing interest asynchronously: /a/b/c");
+ Client client = new Client();
+ ClientObserver observer = client.get(face, new Name("/a/b/c"));
+
+ // wait
+ while (observer.responses() == 0) {
+ Thread.sleep(10);
+ }
+ Data data = (Data) observer.getFirst().getPacket();
+ assertEquals(new Blob("...").buf(), data.getContent().buf());
+ }
}
diff --git a/src/test/java/com/intel/jndn/utils/PublisherTest.java b/src/test/java/com/intel/jndn/utils/PublisherTest.java
new file mode 100644
index 0000000..3c780ef
--- /dev/null
+++ b/src/test/java/com/intel/jndn/utils/PublisherTest.java
@@ -0,0 +1,120 @@
+/*
+ * File name: PublisherTest.java
+ *
+ * Purpose:
+ *
+ * © Copyright Intel Corporation. All rights reserved.
+ * Intel Corporation, 2200 Mission College Boulevard,
+ * Santa Clara, CA 95052-8119, USA
+ */
+package com.intel.jndn.utils;
+
+import com.intel.jndn.mock.MockTransport;
+import java.io.IOException;
+import net.named_data.jndn.Data;
+import net.named_data.jndn.Face;
+import net.named_data.jndn.Name;
+import net.named_data.jndn.encoding.EncodingException;
+import net.named_data.jndn.encoding.tlv.Tlv;
+import net.named_data.jndn.encoding.tlv.TlvEncoder;
+import net.named_data.jndn.security.SecurityException;
+import net.named_data.jndn.util.Blob;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+/**
+ *
+ * @author Andrew Brown <andrew.brown@intel.com>
+ */
+public class PublisherTest {
+
+ /**
+ * Test of publish method, of class Publisher.
+ */
+ @Test
+ public void testPublishSubscribe() throws IOException, EncodingException, InterruptedException, SecurityException {
+ MockTransport transport = new MockTransport();
+ Face face = new Face(transport, null);
+ final Counter counter = new Counter();
+
+ // setup subscriber
+ Subscriber subscriber = new Subscriber(face, new Name("/test/channel"));
+ subscriber.addType("example", TestExample.class);
+ subscriber.on(TestExample.class, new Subscriber.OnPublish<TestExample>() {
+ @Override
+ public void onPublish(TestExample publishedObject) {
+ counter.inc();
+ assertEquals(1, publishedObject.a);
+ assertEquals(true, publishedObject.b);
+ }
+ });
+
+ // setup publisher
+ Publisher publisher = new Publisher(face, new Name("/test/channel"), false);
+ publisher.addType("example", TestExample.class);
+ publisher.publish(new TestExample(1, true));
+
+ // process events
+ while(counter.get() == 0){
+ Thread.sleep(10);
+ }
+
+ // check
+ assertEquals(1, counter.get());
+ }
+
+ public Data fakeManagementSuccess(Name forName, int statusCode, String statusText){
+ TlvEncoder encoder = new TlvEncoder(1500);
+ int saveLength = encoder.getLength();
+
+ // encode backwards
+ encoder.writeBlobTlv(Tlv.NfdCommand_StatusText, new Blob(statusText).buf());
+ encoder.writeNonNegativeIntegerTlv(Tlv.NfdCommand_StatusCode, statusCode);
+ encoder.writeTypeAndLength(Tlv.NfdCommand_ControlResponse, encoder.getLength() - saveLength);
+ Blob content = new Blob(encoder.getOutput(), false);
+
+ // now create data packet
+ Data data = new Data(forName);
+ data.setContent(content);
+ return data;
+ }
+
+ class TestExample {
+
+ int a;
+ boolean b;
+
+ public TestExample(int a, boolean b) {
+ this.a = a;
+ this.b = b;
+ }
+ }
+
+ /**
+ * Count reference
+ */
+ class Counter {
+
+ int count = 0;
+
+ public void inc() {
+ count++;
+ }
+
+ public int get() {
+ return count;
+ }
+ }
+
+// /**
+// * Test of clearOldPublished method, of class Publisher.
+// */
+// @Test
+// public void testClearOldPublished() {
+// System.out.println("clearOldPublished");
+// Publisher instance = null;
+// instance.clearOldPublished();
+// // TODO review the generated test code and remove the default call to fail.
+// fail("The test case is a prototype.");
+// }
+}