<commit_before>/**
 * Part of the CCNx Java Library.
 *
 * Copyright (C) 2008, 2009 Palo Alto Research Center, Inc.
 *
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License version 2.1
 * as published by the Free Software Foundation. 
 * This library is distributed in the hope that 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. You should have received
 * a copy of the GNU Lesser General Public License along with this library;
 * if not, write to the Free Software Foundation, Inc., 51 Franklin Street,
 * Fifth Floor, Boston, MA 02110-1301 USA.
 */

package org.ccnx.ccn.impl;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.ccnx.ccn.CCNFilterListener;
import org.ccnx.ccn.CCNInterestListener;
import org.ccnx.ccn.KeyManager;
import org.ccnx.ccn.config.SystemConfiguration;
import org.ccnx.ccn.impl.InterestTable.Entry;
import org.ccnx.ccn.impl.support.Log;
import org.ccnx.ccn.io.content.ContentEncodingException;
import org.ccnx.ccn.protocol.ContentName;
import org.ccnx.ccn.protocol.ContentObject;
import org.ccnx.ccn.protocol.Interest;
import org.ccnx.ccn.protocol.WirePacket;


/**
 * The low level interface to ccnd. Connects to a ccnd and maintains the connection by sending 
 * heartbeats to it.  Other functions include reading and writing interests and content
 * to/from the ccnd, starting handler threads to feed interests and content to registered handlers,
 * and refreshing unsatisfied interests. 
 * 
 * This class attempts to notice when a ccnd has died and to reconnect to a ccnd when it is restarted.
 * 
 * It also handles the low level output "tap" functionality - this allows inspection or logging of
 * all the communications with ccnd.
 * 
 * Starts a separate thread to listen to, decode and handle incoming data from ccnd.
 */
public class CCNNetworkManager implements Runnable {
	
	public static final int DEFAULT_AGENT_PORT = 9695; // ccnx registered port
	public static final String DEFAULT_AGENT_HOST = "localhost";
	public static final String PROP_AGENT_PORT = "ccn.agent.port";
	public static final String PROP_AGENT_HOST = "ccn.agent.host";
	public static final String PROP_TAP = "ccn.tap";
	public static final String ENV_TAP = "CCN_TAP"; // match C library
	public static final int MAX_PAYLOAD = 8800; // number of bytes in UDP payload
	public static final int SOCKET_TIMEOUT = 1000; // period to wait in ms.
	public static final int PERIOD = 2000; // period for occasional ops in ms.
	public static final int HEARTBEAT_PERIOD = 3500;
	public static final int MAX_PERIOD = PERIOD * 8;
	public static final String KEEPALIVE_NAME = "/HereIAm";
	public static final int THREAD_LIFE = 8;	// in seconds
	
	/*
	 * Static singleton.
	 */
	
	protected Thread _thread = null; // the main processing thread
	protected ThreadPoolExecutor _threadpool = null; // pool service for callback threads
	protected DatagramChannel _channel = null; // for use by run thread only!
	protected Selector _selector = null;
	protected Throwable _error = null; // Marks error state of socket
	protected boolean _run = true;
	protected KeyManager _keyManager;
	protected ContentObject _keepalive; 
	protected FileOutputStream _tapStreamOut = null;
	protected FileOutputStream _tapStreamIn = null;
	protected long _lastHeartbeat = 0;
	protected int _port = DEFAULT_AGENT_PORT;
	protected String _host = DEFAULT_AGENT_HOST;

	// Tables of interests/filters: users must synchronize on collection
	protected InterestTable<InterestRegistration> _myInterests = new InterestTable<InterestRegistration>();
	protected InterestTable<Filter> _myFilters = new InterestTable<Filter>();
	
	private Timer _periodicTimer = null;
	private boolean _timersSetup = false;
	
	/**
	 * Do scheduled writes of heartbeats and interest refreshes
	 */
	private class PeriodicWriter extends TimerTask {
		// TODO Interest refresh time is supposed to "decay" over time but there are currently
		// unresolved problems with this.
		public void run() {
			
			long ourTime = new Date().getTime();
			if ((ourTime - _lastHeartbeat) > HEARTBEAT_PERIOD) {
				_lastHeartbeat = ourTime;
				heartbeat();
			}

			long minRefreshTime = PERIOD + ourTime;
			// Library.finest("Refreshing interests (size " + _myInterests.size() + ")");

			// Re-express interests that need to be re-expressed
			try {
				synchronized (_myInterests) {
					for (Entry<InterestRegistration> entry : _myInterests.values()) {
						InterestRegistration reg = entry.value();
						if (ourTime > reg.nextRefresh) {
							Log.finer("Refresh interest: {0}", reg.interest);
							// Temporarily back out refresh period decay
							//reg.nextRefreshPeriod = (reg.nextRefreshPeriod * 2) > MAX_PERIOD ? MAX_PERIOD
									//: reg.nextRefreshPeriod * 2;
							reg.nextRefresh += reg.nextRefreshPeriod;
							try {
								write(reg.interest);
							} catch (NotYetConnectedException nyce) {}
						}
						if(minRefreshTime > reg.nextRefresh)
							minRefreshTime = reg.nextRefresh;
					}
				}
			} catch (ContentEncodingException xmlex) {
				Log.severe("Processing thread failure (Malformed datagram): {0}", xmlex.getMessage()); 
				Log.warningStackTrace(xmlex);
			}
			
			long checkDelay = minRefreshTime - System.currentTimeMillis();
			if(checkDelay < 0)
				checkDelay = 0;
			if(checkDelay > PERIOD)
				checkDelay = PERIOD;
			
			_periodicTimer.schedule(new PeriodicWriter(), checkDelay);
		}
	}
	
	/**
	 * Send the heartbeat. Also attempt to detect ccnd going down.
	 */
	private void heartbeat() {
		try {
			ByteBuffer heartbeat = ByteBuffer.allocate(1);
			if (_channel.isConnected())
				_channel.write(heartbeat);
		} catch (IOException io) {
			// We do not see errors on send typically even if 
			// agent is gone, so log each but do not track
			Log.warning("Error sending heartbeat packet: {0}", io.getMessage());
			try {<commit_msg>fix format<commit_after>/**
 * Part of the CCNx Java Library.
 *
 * Copyright (C) 2008, 2009 Palo Alto Research Center, Inc.
 *
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License version 2.1
 * as published by the Free Software Foundation. 
 * This library is distributed in the hope that 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. You should have received
 * a copy of the GNU Lesser General Public License along with this library;
 * if not, write to the Free Software Foundation, Inc., 51 Franklin Street,
 * Fifth Floor, Boston, MA 02110-1301 USA.
 */

package org.ccnx.ccn.impl;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.NotYetConnectedException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Date;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.ccnx.ccn.CCNFilterListener;
import org.ccnx.ccn.CCNInterestListener;
import org.ccnx.ccn.KeyManager;
import org.ccnx.ccn.config.SystemConfiguration;
import org.ccnx.ccn.impl.InterestTable.Entry;
import org.ccnx.ccn.impl.support.Log;
import org.ccnx.ccn.io.content.ContentEncodingException;
import org.ccnx.ccn.protocol.ContentName;
import org.ccnx.ccn.protocol.ContentObject;
import org.ccnx.ccn.protocol.Interest;
import org.ccnx.ccn.protocol.WirePacket;


/**
 * The low level interface to ccnd. Connects to a ccnd and maintains the connection by sending 
 * heartbeats to it.  Other functions include reading and writing interests and content
 * to/from the ccnd, starting handler threads to feed interests and content to registered handlers,
 * and refreshing unsatisfied interests. 
 * 
 * This class attempts to notice when a ccnd has died and to reconnect to a ccnd when it is restarted.
 * 
 * It also handles the low level output "tap" functionality - this allows inspection or logging of
 * all the communications with ccnd.
 * 
 * Starts a separate thread to listen to, decode and handle incoming data from ccnd.
 */
public class CCNNetworkManager implements Runnable {
	
	public static final int DEFAULT_AGENT_PORT = 9695; // ccnx registered port
	public static final String DEFAULT_AGENT_HOST = "localhost";
	public static final String PROP_AGENT_PORT = "ccn.agent.port";
	public static final String PROP_AGENT_HOST = "ccn.agent.host";
	public static final String PROP_TAP = "ccn.tap";
	public static final String ENV_TAP = "CCN_TAP"; // match C library
	public static final int MAX_PAYLOAD = 8800; // number of bytes in UDP payload
	public static final int SOCKET_TIMEOUT = 1000; // period to wait in ms.
	public static final int PERIOD = 2000; // period for occasional ops in ms.
	public static final int HEARTBEAT_PERIOD = 3500;
	public static final int MAX_PERIOD = PERIOD * 8;
	public static final String KEEPALIVE_NAME = "/HereIAm";
	public static final int THREAD_LIFE = 8;	// in seconds
	
	/*
	 * Static singleton.
	 */
	
	protected Thread _thread = null; // the main processing thread
	protected ThreadPoolExecutor _threadpool = null; // pool service for callback threads
	protected DatagramChannel _channel = null; // for use by run thread only!
	protected Selector _selector = null;
	protected Throwable _error = null; // Marks error state of socket
	protected boolean _run = true;
	protected KeyManager _keyManager;
	protected ContentObject _keepalive; 
	protected FileOutputStream _tapStreamOut = null;
	protected FileOutputStream _tapStreamIn = null;
	protected long _lastHeartbeat = 0;
	protected int _port = DEFAULT_AGENT_PORT;
	protected String _host = DEFAULT_AGENT_HOST;

	// Tables of interests/filters: users must synchronize on collection
	protected InterestTable<InterestRegistration> _myInterests = new InterestTable<InterestRegistration>();
	protected InterestTable<Filter> _myFilters = new InterestTable<Filter>();
	
	private Timer _periodicTimer = null;
	private boolean _timersSetup = false;
	
	/**
	 * Do scheduled writes of heartbeats and interest refreshes
	 */
	private class PeriodicWriter extends TimerTask {
		// TODO Interest refresh time is supposed to "decay" over time but there are currently
		// unresolved problems with this.
		public void run() {
			
			long ourTime = new Date().getTime();
			if ((ourTime - _lastHeartbeat) > HEARTBEAT_PERIOD) {
				_lastHeartbeat = ourTime;
				heartbeat();
			}

			long minRefreshTime = PERIOD + ourTime;
			// Library.finest("Refreshing interests (size " + _myInterests.size() + ")");

			// Re-express interests that need to be re-expressed
			try {
				synchronized (_myInterests) {
					for (Entry<InterestRegistration> entry : _myInterests.values()) {
						InterestRegistration reg = entry.value();
						if (ourTime > reg.nextRefresh) {
							Log.finer("Refresh interest: {0}", reg.interest);
							// Temporarily back out refresh period decay
							//reg.nextRefreshPeriod = (reg.nextRefreshPeriod * 2) > MAX_PERIOD ? MAX_PERIOD
									//: reg.nextRefreshPeriod * 2;
							reg.nextRefresh += reg.nextRefreshPeriod;
							try {
								write(reg.interest);
							} catch (NotYetConnectedException nyce) {}
						}
						if (minRefreshTime > reg.nextRefresh)
							minRefreshTime = reg.nextRefresh;
					}
				}
			} catch (ContentEncodingException xmlex) {
				Log.severe("Processing thread failure (Malformed datagram): {0}", xmlex.getMessage()); 
				Log.warningStackTrace(xmlex);
			}
			
			long checkDelay = minRefreshTime - System.currentTimeMillis();
			if (checkDelay < 0)
				checkDelay = 0;
			if (checkDelay > PERIOD)
				checkDelay = PERIOD;
			
			_periodicTimer.schedule(new PeriodicWriter(), checkDelay);
		}
	}
	
	/**
	 * Send the heartbeat. Also attempt to detect ccnd going down.
	 */
	private void heartbeat() {
		try {
			ByteBuffer heartbeat = ByteBuffer.allocate(1);
			if (_channel.isConnected())
				_channel.write(heartbeat);
		} catch (IOException io) {
			// We do not see errors on send typically even if 
			// agent is gone, so log each but do not track
			Log.warning("Error sending heartbeat packet: {0}", io.getMessage());
			try {
{'fraction_non_alphanumeric': 0.08369977255496588, 'fraction_numerical': 0.007884761182714178, 'mean_word_length': 3.972107048624199, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4327722', 'n_tokens_mistral': 4242, 'n_tokens_neox': 3693, 'n_words': 1330}