

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of PSYCLONE version 2.0.0.170915_Beta4 (15 September 2017)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2017
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  PSYCLONE 2.0 is released under the GNU Lesser General Public License 
//_/_/  (LGPL) version 3.0, and is subject to the following
//_/_/  
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/    
//_/_/  A commercial license for PSYCLONE 2.0 is also available to 
//_/_/  any interested party from the copyright holder.
//_/_/  
//_/_/  You can redistribute this copy of PSYCLONE 2.0 and/or modify it under 
//_/_/  the terms of the GNU General Public License as published by the Free 
//_/_/  Software Foundation, either version 3 of the License, or (at your 
//_/_/  option) any later version.
//_/_/  
//_/_/  You should have received a copy of the GNU General Public License
//_/_/  and the GNU Lesser General Public License along with PSYCLONE 2.0.  
//_/_/  If not, see <http://www.gnu.org/licenses/>.
//_/_/  
//_/_/  PSYCLONE 2.0 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 General Public License for more details.
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#if !defined(_NETWORKING_H_)
#define _NETWORKING_H_

#pragma once

#include "PsyAPI.h"
#include "NetworkManager.h"
#include "DataMessage.h"
#include "MovingAverage.h"
#include "ServiceManager.h"

// Built-in message types

#define NODEMAP				GenObjID(	101	)
#define NODESTATUSMAP		GenObjID(	102	)

#define NODE_TIMENODE	1
#define NODE_REQUIRED	2
#define NODE_OPTIONAL	3
#define NODE_EXTERNAL	4

#define SYNC_RESERVE_ID			1
#define SYNC_TYPELEVEL_ID		2
#define SYNC_CONTEXTLEVEL_ID	3
#define SYNC_CRANK_ID			4
#define SYNC_COMPONENT_ID		5
#define SYNC_TAG_ID			6
#define SYNC_SUBSCRIPTION		7
#define SYNC_COMPONENT_LOC		8
#define SYNC_RESERVE_IDS		10
#define BULK_SYNC_TYPELEVEL_ID		11
#define BULK_SYNC_CONTEXTLEVEL_ID	12
#define BULK_SYNC_CRANK_ID			13
#define BULK_SYNC_COMPONENT_ID		14
#define BULK_SYNC_TAG_ID			15
#define BULK_SYNC_SUBSCRIPTION		16
#define BULK_SYNC_CONTEXTS			17

static const char *SYNCTYPES[] = { 
   "",
   "IDRange",
   "TypeLevelID",
   "ContextLevelID",
   "CrankID",
   "ComponentID",
   "TagID"
};

#define NODESTATUS_ERROR	0
#define NODESTATUS_IDLE		1
#define NODESTATUS_INIT		2
#define NODESTATUS_TIMESYNC	3
#define NODESTATUS_MAPPING	5
#define NODESTATUS_NETREADY	6
#define NODESTATUS_IDSYNC	7
#define NODESTATUS_SPEC		8
#define NODESTATUS_DATASYNC	9
#define NODESTATUS_READY	10

#define CONNECTION_ERROR			0
#define CONNECTION_IDLE				1
#define CONNECTION_DISCONNECTED		2
#define CONNECTION_RETRYING			3
#define CONNECTION_INITIALISING		4
#define CONNECTION_READY			5

namespace	cmlabs{



struct TypeLevelID {
	uint32 size;
	uint32 cid;
	uint16 id; // id, corresponds to the index in map
	uint64 createdTime; // timestamp when entered
	char name[MAXKEYNAMELEN+1]; // name in text
};

struct ContextLevelID {
	uint32 size;
	uint32 cid;
	uint16 id; // id, corresponds to the index in map
	uint64 createdTime; // timestamp when entered
	char name[MAXKEYNAMELEN+1]; // name in text
};

struct TagID {
	uint32 size;
	uint32 cid;
	uint16 id; // id, corresponds to the index in map
	uint64 createdTime; // timestamp when entered
	char name[MAXKEYNAMELEN+1]; // name in text
};

struct CrankID {
	uint32 size;
	uint32 cid;
	uint16 id; // id, corresponds to the index in map
	uint32 compID; // parent component id
	uint64 createdTime; // timestamp when entered
	char name[MAXKEYNAMELEN+1]; // name in text
	char language[MAXKEYNAMELEN + 1]; // language in text
	char function[MAXKEYNAMELEN+1]; // name in text
	char libraryFilename[MAXKEYNAMELEN+1]; // name in text
	char script[MAXSCRIPTLEN+1]; // name in text
};

struct ComponentID {
	uint32 size;
	uint32 cid;
	uint32 id;
	uint8 type; // id of node component is running on currently
	uint8 policy; // migration policy
	uint8 selfTrigger; // selftriggering policy
	uint16 nodeID; // id of node component is running on currently
	uint16 procID; // id of node component is running on currently
	uint64 createdTime;
	char name[MAXKEYNAMELEN]; // name in text
};

class Node;

			// ###### Merge Entries/Status/Cons

struct NodeMap {
	uint32	size;
	uint32	cid;
	uint64	time;
	uint64	instanceID; // auto incremented number
	uint16	timesyncID;
	uint32	count; // count of all nodes incl. local node entry
	uint16	nodeID; // local node id = entry id
	uint16	nodeStatus;
	// NodeEntry * count
};

#define MAXENTRYIPADDRESSCOUNT 64
struct NodeEntry {
	// Global information about a node
	uint16 nodeID;
	uint64 instanceID;
	uint16 type;
	uint16 status;
	uint32 ipAddressCount;
	uint32 ipAddresses[MAXENTRYIPADDRESSCOUNT];
	uint16 port;
	char name[256];
	char addr[256];
	char hostname[256];

	// Local node data used locally
	uint16 timesyncNodeID;  // Who do I synchronise time with
	uint16 configNodeID;    // ID of the central config node (during initial setup)
	uint16 configNodeViaID; // If not connected to config node directly, who do I go via

	// About connection to other node
	uint64 location;        // On which address do I connect to this node
	uint16 constatus;       // Connection status to this node
	uint32 timeout;         // Connection timeout to this node
	uint64 conid;           // Internal conid for my TCP connection to this node
	uint64 udpConid;        // Internal conid for my UDP connection to this node
	uint64 lastContact;     // When did I last speak to this node
	uint64 msgtime;         // Used for timesync and ping
	double dataInBytes[4];  // Stats {1,10,60,All} sec to this node
	double dataOutBytes[4]; // Stats {1,10,60,All} sec to this node

	// utils functions
	uint32 addIPAddresses(uint32* newAddresses, uint32 count);
};

struct TimeSyncData {
	uint32	size;
	uint32	cid;
	uint64	t1;
	uint64	t2;
	int64	total;
	int64	msgtime;
	uint32	count;
};

class ReplyCheckEntry {
public:
	ReplyCheckEntry() {id = 0;};
	~ReplyCheckEntry();
	uint8 getStatus();
	uint8 getReplyStatus(uint32& total, uint32& success, uint32& failed);
	uint32 id;
	std::map<uint16, uint8> entries;
	std::map<uint16, DataMessage*> messages;
	utils::Mutex mutex;
	utils::Semaphore semaphore;
};

class ReplyCheck {
public:
	ReplyCheck() {lastID = 0;};
	~ReplyCheck();
	uint32 newEntry();
	bool addReplier(uint32 id, uint16 replier);
	bool addReply(uint32 id, uint16 replier, bool answer, DataMessage* msg = NULL);
	uint8 waitForReplies(uint32 id, uint32 timeout, std::map<uint16, DataMessage*>* map = NULL, bool deleteOnTimeout = true);
	uint8 getReplyStatus(uint32 id, uint32& total, uint32& success, uint32& failed);
private:
	uint32 lastID;
	std::map<uint32, ReplyCheckEntry*> replyEntries;
};

//  <interface name="MyWebInterface" node="Node0" port="1234" protocol="HTTP" service="MyWebService" timeout="3000" default="yes" />

#define INTERFACE_MESSAGE	0
#define INTERFACE_TELNET	1
#define INTERFACE_HTTP		2
#define INTERFACE_RAW		3

#define INTERFACE_PROTO_TCP	0
#define INTERFACE_PROTO_UDP	1

class Authentication {
public:
	Authentication();
	~Authentication();

	bool fromXML(XMLNode &n);
};

class Interface {
public:
	Interface();
	~Interface();

	bool fromXML(XMLNode &n);

	uint16 id;
	uint16 port;
	uint8 protocol;
	uint8 encryption;
	uint32 timeout;
	uint8 ipproto;
	bool isDefault;
	std::string name;
	std::string nodeName;
	std::string serviceName;
	uint32 serviceID;
	Authentication* auth;
};

struct InterfacePack {
	Interface* interfaces[4];
};

#define QUERYNODE_GETINFOPACK				1
#define QUERYNODE_GETCOMPDATA				2
#define QUERYNODE_SETCOMPDATA				3
#define QUERYNODE_GETCOMPPRIVATEDATA		4
#define QUERYNODE_GETMESSAGECOPY			5
#define QUERYNODE_GETPERFMATRICES			6
#define QUERYNODE_GETCOMPPRIVATEKEYSTYPES	7

class Node;
//class Networking : public NetworkConnectionReceiver, public NetworkDataReceiver, public Runnable {
class Networking : public NetworkReceiver, public Runnable {
	friend class Node;
	friend class PsyProbe;
public:
	friend THREAD_RET THREAD_FUNCTION_CALL NodeNetworkingRun(THREAD_ARG arg);

	static const uint8 NetworkProtocols[];

	static struct PsyType CTRL_NODE_WELCOME;
	static struct PsyType CTRL_NODE_JOIN;

	static struct PsyType CTRL_NODE_TIMESYNC;
	static struct PsyType CTRL_NODE_TIMESYNC_REPLY;
	static struct PsyType CTRL_NODE_NODEMAP;
	static struct PsyType CTRL_NODE_NODEMAP_REQUEST;
	static struct PsyType CTRL_NODE_STATUSMAP;
	static struct PsyType CTRL_NODE_PERFSYNC;
	
	static struct PsyType CTRL_NODE_SYNC_REQ;
	static struct PsyType CTRL_NODE_SYNC_REPLY;
	static struct PsyType CTRL_NODE_CONFIG;
	static struct PsyType CTRL_NODE_CONFIG_SUCCESS;
	static struct PsyType CTRL_NODE_CONFIG_FAILED;
	static struct PsyType CTRL_NODE_SYNC_ID;
	static struct PsyType CTRL_NODE_SYNC_ID_SUCCESS;
	static struct PsyType CTRL_NODE_SYNC_ID_FAILED;
	static struct PsyType CTRL_NODE_CONFIRM_ID;
	static struct PsyType CTRL_NODE_CONFIRM_ID_SUCCESS;
	static struct PsyType CTRL_NODE_CONFIRM_ID_FAILED;
	static struct PsyType CTRL_NODE_CANCEL_ID;
	static struct PsyType CTRL_NODE_CANCEL_ID_SUCCESS;
	static struct PsyType CTRL_NODE_CANCEL_ID_FAILED;
	static struct PsyType CTRL_NODE_CONTEXT_ACTIVE;
	static struct PsyType CTRL_NODE_SIGNAL;
	static struct PsyType CTRL_NODE_QUERY;
	static struct PsyType CTRL_NODE_QUERY_SUCCESS;
	static struct PsyType CTRL_NODE_QUERY_FAILED;
	static struct PsyType CTRL_NODE_SYNC_COMPONENT;
	static struct PsyType CTRL_NODE_SYNC_COMPONENT_SUCCESS;
	static struct PsyType CTRL_NODE_SYNC_COMPONENT_FAILED;
	static struct PsyType CTRL_NODE_SYNC_SUBSCRIPTION;
	static struct PsyType CTRL_NODE_SYNC_SUBSCRIPTION_SUCCESS;
	static struct PsyType CTRL_NODE_SYNC_SUBSCRIPTION_FAILED;
	static struct PsyType CTRL_PSYPROBE_CREATECUSTOMPAGE;
	static struct PsyType CTRL_PSYPROBE_CREATECUSTOMPAGE_SUCCESS;
	static struct PsyType CTRL_PSYPROBE_CREATECUSTOMPAGE_FAILED;


	Networking(Node* node);
	~Networking();
	bool init(uint16 port, ServiceManager* serviceManager);
	bool shutdown();

	bool addInterface(XMLNode &n);
	bool addInterface(const char* name, uint16 port, uint8 ipproto, uint8 encryption, uint8 protocol, uint32 protocolTimeout, bool isDefaultProtocol, const char* nodeName, const char* serviceName);
	bool addInterface(Interface* iface);
	bool removeInterface(uint16 port, uint8 protocol);

	bool setupNodeNetwork(uint16 mainPort, XMLNode& mainNode);
	bool waitForNodeNetwork(uint32 ms = 0);
	uint16 connectToNodeNetwork(const char* otherNodeAddress, uint16 otherNodePort);
	bool updateNodeNetwork(uint16 remoteNodeID, NodeMap* map, uint64 conid);

	uint16 getLocalNodeStatus();
	bool setLocalNodeStatus(uint16 newStatus, bool onlyIfLarger = false);
	uint16 getNodeIssues(std::list<NodeEntry> &nodeIssues);

	bool getNodeID(const char* name, uint16& id);
	std::string getNodeName(uint16 id);
	bool calcViaNodeID(uint16 id, uint16& viaID);

	//bool addTCPPort(uint16 port, uint8 encryption, uint8 protocol, bool isAsync, uint32 protocolTimeout, bool isDefaultProtocol);
	//bool addUDPPort(uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect);
	//bool removeTCPPort(uint16 port, uint8 protocol);
	//bool removeUDPPort(uint16 port);
	
	bool receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid);
	bool receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid);
	bool receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid);
	bool receiveTelnetLine(TelnetLine* line, NetworkChannel* channel, uint64 conid);
	bool receiveHTTPReply(HTTPReply* reply, HTTPRequest* req, NetworkChannel* channel, uint64 conid);

	uint16 getNodeCount();
	NodeEntry* getNodeEntry(uint16 nodeID);

	uint64 openConnectionToHost(uint32 host, uint16 port);
	
	bool sendMessage(DataMessage* msg, bool deleteWhenDone = true);
	bool sendMessageToHost(DataMessage* msg, uint32 host, uint16 port);
	bool sendMessageToConnection(DataMessage* msg, uint64 conid);
	bool sendMessageToNode(uint16 id, DataMessage* msg, bool deleteWhenDone = true);
	bool sendMessageToNode(uint16 id, PsyType controltype);
	uint16 sendMessageToAllOtherNodes(PsyType controltype);
	uint16 sendMessageToNodes(uint16* ids, uint16 count, DataMessage* msg, bool deleteWhenDone = true);
	uint16 sendMessageToAllOtherNodes(DataMessage* msg, bool deleteWhenDone = true);
	uint8 syncNodes(DataMessage* msg, uint32 timeout, bool deleteWhenDone, std::map<uint16, DataMessage*>* replyMap = NULL);
	uint8 syncNode(uint16 nodeID, DataMessage* msg, uint32 timeout, bool deleteWhenDone, std::map<uint16, DataMessage*>* replyMap = NULL);
	uint8 queryNode(uint16 otherNodeID, DataMessage* msg, uint32 timeout, bool deleteWhenDone, DataMessage** replyMsg = NULL);

	DataMessage* waitForIncomingMessage(uint32 timeout);

	NetworkManager* manager;
	uint16 localNodeID;

protected:

	uint32 findServiceID(uint8 type, NetworkChannel* channel, uint64 conid);

	uint32 lastInterfaceID;
	std::map<NetworkChannel*, uint16> channelPorts;
	std::map<uint16, InterfacePack> portInterfacePacks;

	NodeEntry* localNodeEntry;
	std::map<uint16, NodeEntry*> nodeEntries;
	utils::Mutex nodeEntriesMutex;
	std::map<uint16, NodeMap*> nodeMaps;
	utils::Mutex nodeMapsMutex;

	std::queue<DataMessage*> inputMessageQueue;
	utils::Semaphore inputMessageQueueSemaphore;
	std::queue<DataMessage*> outputMessageQueue;
	utils::Semaphore outputMessageQueueSemaphore;

	// Statistics
	std::map<uint16, MovingAverage*> nodeInStats;
	std::map<uint16, MovingAverage*> nodeOutStats;

	std::map<uint64, uint64> externalConnections;

	// TimeSync
	int64 timeAdjustRemain;
	int64 timeStepSize;
	uint16 timeSyncCount;
	bool startTimesync(uint32 n = 1, uint64 total = 0, uint64 msgtime = 0);
	bool adjustTimesync(int64 amount);

	ReplyCheck replyCheck;

	uint16 altLocalPort;
	char localNodeName[256];
	uint16 timesyncNodeID;
	uint32 heartbeatInterval;
	uint32 timesyncInterval;
	uint32 timestepInterval;
	uint32 pingInterval;
	uint32 conCheckInterval;
	uint32 reqNodeTimeout;
	ServiceManager* serviceManager;
	Node* node;

	DataMessage* createHeartbeatMessage();
	bool sendHeartbeat();

	bool waitForAllNodesStatus(uint16 status, uint32 ms = 0);

	DataMessage* waitForOutgoingMessage(uint32 timeout);

	bool checkNodeInstanceID(uint16 nodeID, uint64 instID, uint64 conID);
	char* generateNodeSyncData(uint32& size);

	uint16 getAvailableNodeID();
	bool addToConnectionStats(uint16 id, uint32 inBytes, uint32 outBytes, uint64 time);
	bool updateConnectionStats(NodeEntry* entry);
	bool updateAllConnectionStats();
	bool run();

};


THREAD_RET THREAD_FUNCTION_CALL NodeNetworkingRun(THREAD_ARG arg);

}

#include "Node.h"

#endif // _NODE_H_
