package com.tipcj.demos;

import java.nio.ByteBuffer;

import com.tipcj.*;

public class Tipc_Java_Api_Client {
	public int[] pfds;
	public short[] events;
	public Integer[] revents;
	public TipcSrvConn topSrv;
	public Integer up;
	public TipcAddr srvAddr;
	public TipcConn rdmConn;
	public TipcConn streamConn;
	public TipcConn seqpacketConn;

	public long rdm_service_demo() {
		byte[] inbuf = new byte[Common.BUF_SIZE];
		ByteBuffer bytebuffer = ByteBuffer.allocateDirect(Common.BUF_SIZE);
		byte[] msg = new String("Hello World!!!").getBytes();
		int ret;
		Integer err = new Integer(0);

		if (up == 0) {
			System.out.println("Service on SOCK_RDM went down");
			return 0;
		}
		System.out.println("-------------------------------------");
		System.out.println("Service on SOCK_RDM came up");
		System.out.println("Sending msg: " + new String(msg) + " on SOCK_RDM --> " + srvAddr);

		if (rdmConn.Sock_Rejectable() < 0) {
			System.err.println("Set rejectable failed");
			return 0;
		}

		if (rdmConn.Sendto(msg, srvAddr) != msg.length) {
			System.err.println("sendto() failed");
			return 0;
		}

		ret = rdmConn.Recvfrom(bytebuffer, bytebuffer.limit(), srvAddr, null, err);
		if ((ret < 0) || (err != 0)) {
			System.err.println("Unexpected response");
			return 0;
		}

		bytebuffer.get(inbuf, 0, ret);
		System.out.println("Received response: " + new String(inbuf) + " <--" + srvAddr);
		return srvAddr.Node;
	}

	public void rdm_reject_demo(long srv_node) {
		TipcAddr invalid = new TipcAddr(42, 1, srv_node);
		byte[] msg = new String("Hello World").getBytes();
		byte[] inbuf = new byte[Common.BUF_SIZE];
		ByteBuffer bytebuffer = ByteBuffer.allocateDirect(Common.BUF_SIZE);
		Integer err = new Integer(0);
		int ret;

		if (up == 0) {
			return;
		}

		System.out.println("Sending msg: " + new String(msg) + " on SOCK_RDM --> " + invalid + " (non-existing)");
		if (rdmConn.Sendto(msg, invalid) != msg.length) {
			System.out.println("Client sendto() failed: No route to host");
			return;
		}

		ret = rdmConn.Recvfrom(bytebuffer, bytebuffer.limit(), srvAddr, null, err);
		if ((ret < 0) || (err == 0)) {
			System.err.println(String.format("Unexpected response err:%d", err));
		}
		bytebuffer.get(inbuf, 0, ret);
		System.out.println("Received rejected msg: '" + new String(inbuf) + "' <-- " + srvAddr + " err:" + err);
		System.out.println("-------------------------------------");
	}

	public void stream_service_demo() {
		TipcAddr srv = new TipcAddr(Common.STREAM_SRV_TYPE, Common.SRV_INST, 0);
		byte[] msg = new String("Hello World").getBytes();
		byte[] inbuf = new byte[Common.BUF_SIZE];
		int ret = 0;

		if (up == 0) {
			System.out.println("Service on SOCK_STREAM went down");
			return;
		}
		System.out.println("-------------------------------------");
		System.out.println("Service on SOCK_STREAM came up");
		System.out.println("Performing implicit two-way connect with message '" + new String(msg) + "' --> " + srv);

		try {
				Thread.sleep(1000);
		} catch (InterruptedException ex) {}

		ret = streamConn.Sendto(msg, srv);
		if (ret != msg.length) {
			System.err.println("send() failed");
		}

		System.err.println("Receiving ...");
		ret = streamConn.Recv(inbuf, false);
		if (ret <= 0) {
			System.err.println("Unexpected response");
			return;
		}

		System.out.println("Received response: " + new String(inbuf) + " on SOCK_STREAM connection");
		System.out.println("SOCK_STREAM connection established --> " + srvAddr);
		System.out.println("-------------------------------------");
	}

	public void seqpacket_service_demo() {
		TipcAddr srv = new TipcAddr(Common.SEQPKT_SRV_TYPE, Common.SRV_INST, 0);
		byte[] msg = new String("Hello World").getBytes();
		byte[] inbuf = new byte[Common.BUF_SIZE];
		int ret = 0;

		if (up == 0) {
			System.out.println("Service on SOCK_SEQPACKET went down");
			return;
		}

		System.out.println("-------------------------------------");
		System.out.println("Service on SOCK_SEQPACKET came up");

		System.out.println("Connecting to: -->" + srvAddr);

		ret =  seqpacketConn.Connect(srvAddr);
		if (ret < 0) {
			System.err.println("connect() failed");
			return;
		}

		System.out.println(String.format("Sending msg: %s on connection", new String(msg)));
		ret = seqpacketConn.Send(msg);
		if (ret != msg.length) {
			System.err.println("send() failed");
			return;
		}

		ret = seqpacketConn.Recv(inbuf, true);
		if (ret <= 0) {
			System.err.println("Unexpected response");
			return;
		}

		System.out.println("Received response: " + new String(inbuf) + " on SOCK_SEQPACKET connection");
		System.out.println("-------------------------------------");
	}

	public Tipc_Java_Api_Client() {
		/* Create traffic sockets */
		rdmConn = new TipcConn(SocketType.SOCK_RDM);
		streamConn = new TipcConn(SocketType.SOCK_STREAM);
		seqpacketConn = new TipcConn(SocketType.SOCK_SEQPACKET);

		/* Subscribe for service events */
		topSrv = new TipcSrvConn(0);

		pfds = new int[] {rdmConn.GetFd(), streamConn.GetFd(), seqpacketConn.GetFd(), topSrv.getFd()};
		events = new short[] {Poll.EPOLLIN, Poll.EPOLLIN | Poll.EPOLLHUP, Poll.EPOLLIN | Poll.EPOLLHUP, Poll.EPOLLIN | Poll.EPOLLHUP};
		revents = new Integer[pfds.length];

		srvAddr = new TipcAddr(Common.RDM_SRV_TYPE, Common.SRV_INST, 0);
		up = new Integer(0);

		System.out.println("Waiting for Service " + srvAddr);
		TipcSrvConn.SrvWait(srvAddr, -1);

		if (topSrv.SrvScr(Common.RDM_SRV_TYPE, 0, ~0, false, -1) != 0) {
			System.err.println("subscribe for RDM server failed");
		}

		if (topSrv.SrvScr(Common.STREAM_SRV_TYPE, 0, ~0, false, -1) != 0) {
			System.err.println("subscribe for STREAM server failed");
		}

		if (topSrv.SrvScr(Common.SEQPKT_SRV_TYPE, 0, ~0, false, -1) != 0) {
			System.err.println("subscribe for SEQPACKET server failed");
		}

		while (Poll.pollset(pfds, events, revents, 3000000) >=0) {
			if ((revents[1] & Poll.EPOLLHUP) != 0) {
				System.out.println("SOCK_STREAM connection hangup");
				streamConn.Close();
				streamConn = new TipcConn(SocketType.SOCK_STREAM);
				pfds[1] = streamConn.GetFd();
			}

			if ((revents[2] & Poll.EPOLLHUP) != 0) {
				System.out.println("SOCK_SEQPACKET connection hangup");
				seqpacketConn.Close();
				seqpacketConn = new TipcConn(SocketType.SOCK_SEQPACKET);
				pfds[2] = seqpacketConn.GetFd();
			}

			if ((revents[3] & Poll.EPOLLIN) != 0) {
				if (topSrv.GetEvt(srvAddr, null, up, new Integer(0)) != 0) {
					System.err.println("reception of service event failed");
					break;
				}
				if (srvAddr.Type == Common.RDM_SRV_TYPE) {
					long srv_node = rdm_service_demo();
					rdm_reject_demo(srv_node);
				}

				if (srvAddr.Type == Common.STREAM_SRV_TYPE) {
					stream_service_demo();
				}
				if (srvAddr.Type == Common.SEQPKT_SRV_TYPE) {
					seqpacket_service_demo();
				}
			}
		}
	}

	public static void main(String[] args) {
		System.out.println("****** TIPC JAVA API Demo Client Started ******");
		new Tipc_Java_Api_Client();
		System.out.println("****** TIPC JAVA API Demo Finished ******");
	}

}
