src/main/java/com/intel/jndn/utils/Client.java

/*
 * File name: Client.java
 * 
 * Purpose: Provide a client to simplify information retrieval over the NDN
 * network.
 * 
 * © 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.io.IOException;
import net.named_data.jndn.Data;
import net.named_data.jndn.Face;
import net.named_data.jndn.ForwardingFlags;
import net.named_data.jndn.Interest;
import net.named_data.jndn.Name;
import net.named_data.jndn.OnData;
import net.named_data.jndn.OnInterest;
import net.named_data.jndn.OnRegisterFailed;
import net.named_data.jndn.OnTimeout;
import net.named_data.jndn.encoding.EncodingException;
import net.named_data.jndn.transport.Transport;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 *
 * @author Andrew Brown <andrew.brown@intel.com>
 */
public class Client {

	public static final long DEFAULT_SLEEP_TIME = 20;
	public static final long DEFAULT_TIMEOUT = 2000;
	private static final Logger logger = 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();

		// 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();
		}

		// 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));
	}

	/**
	 * 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());
				}

				// 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();

		// 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));
	}

	/**
	 * 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);

		// 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();
		}

		// return
		logger.debug("Request time (ms): " + (event.getTimestamp() - startTime));
		return (event.isSuccess()) ? (Interest) event.getPacket() : null;
	}

	/**
	 * 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 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);

		// 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);
				}

				// 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();

		return observer;
	}

	/**
	 * 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;
	}
}