Do-It Yourself Multicast Tunneling

package com.cloud.core.net;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.net.SocketException;
import java.nio.charset.Charset;
import java.util.Arrays;

import com.c1as.cloud.core.types.CoreTypes;

/**
 * A very simple multi-cast sender/receiver: IP Multicast is an extension to the standard IP network-level protocol. RFC 1112.
 * IP Multicasting as: "the transmission of an IP datagram to a ‘host group’, a set of zero or more hosts identified by a single IP
 * destination address. A multicast datagram is delivered to all members of its destination host group with the same ‘best-efforts’
 * 
 * <ul>
 * <li>See https://docs.oracle.com/javase/7/docs/api/java/net/MulticastSocket.html
 * <li>https://www.csie.ntu.edu.tw/~course/u1580/Download/Multicast/howipmcworks.html
 * </ul>
 * 
 *
 */
public class NetMulticaster {

    /** Default address */
    public static final String DEFAULT_ADDR = "224.1.2.3";
    
    /** Default port */
    public static final int     DEFAULT_PORT = 6789;

    /** Several standard settings for TTL are specified for the MBONE: 1 for local net, 15 for site, 63 for region and 127 for world. */
    public static final int     DEFAULT_TTL = 15;

    String  address     = DEFAULT_ADDR;
    int     port        = DEFAULT_PORT;
    
    // receive buffer size (8K). The send buffer size cannot exceed this value or extra data will be dropped!.
    int     bufferSize  = 8192;
    
    // The socket
    protected MulticastSocket   s;
    
    // Multi cast group
    protected InetAddress       group;
    
    // 1 thread to receive stuff.
    protected Thread            receiver;
    
    // Listener used to send messages.
    protected MessageListener   listener;
    
    public static interface MessageListener {
        void onMessage (byte[] bytes);
    }
    
    public NetMulticaster()  {
    }

    /**
     * Construct
     * @param address Multicast UDO address.
     * @param port Multicast port.
     */
    public NetMulticaster(String address, int port)  {
        this.address    = address;
        this.port       = port;
    }

    /**
     * Create a {@link MulticastSocket} with the default TTL (15).
     * @throws IOException On socket errors.
     */
    public void open () throws IOException {
        open(DEFAULT_TTL);
    }
    
    /**
     * Create a {@link MulticastSocket} with the given TTL.
     * @param TTL Time-to-live: Several standard settings for TTL are specified for the MBONE: 1 for local net, 15 for site, 63 for region and 127 for world.
     * @throws IOException On socket errors.
     */
    public void open (int TTL) throws IOException {
        s   = new MulticastSocket(port);
        // Several standard settings for TTL are specified for the MBONE: 1 for local net, 15 for site, 63 for region and 127 for world.
        s.setTimeToLive(TTL);
    }
    
    /**
     * Set the multicast arguments. See https://docs.oracle.com/javase/7/docs/api/java/net/MulticastSocket.html
     * @param address Multi-cast address. Default: 224.1.2.3
     * @param port Multi-cast port. Default: 6789
     * @param bufferSize Multi-cast receive buffer size. Default: 2048
     */
    public void configure (String address, int port , int bufferSize) {
        this.address    = address;
        this.port       = port;
        this.bufferSize = bufferSize;
    }

    /**
     * Set the multicast arguments. See https://docs.oracle.com/javase/7/docs/api/java/net/MulticastSocket.html
     * @param address Multi-cast address. Default: 224.1.2.3
     * @param port Multi-cast port. Default: 6789
     */
    public void configure (String address, int port ) {
        configure(address, port, DEFAULT_PORT);
    }

    public void setListener (MessageListener l) {
        listener = l;
    }
    
    /**
     * Join the multicast group given by the default address (224.1.2.3) and port (6789).
     * @throws IOException
     */
    public void joinGroup () throws IOException {
        group   = InetAddress.getByName(address);
        s.joinGroup(group);     
    }
    
    public void shutdown () throws IOException {
        if ( s != null) {
            s.leaveGroup(group);
            s.close();
            s = null;
        }
        
        stopThread();
    }
    
    public boolean isClosed () {
        if ( s == null) {
            return true;
        }
        return s.isClosed();
    }
    
    private void stopThread() {
        if ( receiver == null ) {
            return;
        }
        try {
            receiver.interrupt();
            receiver.join(2000);
        } catch (InterruptedException e) {
        }
    }
    
    /**
     * Send a packet to the multicast group.
     * @param msg Message bytes.
     * @throws IOException On I/O errors.
     */
    public void send (final byte[] msg) throws IOException {
        // guard against someone trying to send when the socket has been shutdown (null)
        if ( s == null) {
            return;
        }
        DatagramPacket hi = new DatagramPacket(msg, msg.length, group, port);
        s.send(hi);
    }
    
    /**
     * Send a text message.
     * @param text Text to send.
     * @throws IOException On I/O errors.
     */
    public void send (final String text) throws IOException {
        byte[] bytes = text.getBytes(Charset.defaultCharset());
        // In UDP there is no reading in chunks . If the size of the message > receive buffer size
        // then data is dropped. See https://stackoverflow.com/questions/15446887/udp-read-data-from-the-queue-in-chunks
        if ( bytes.length > bufferSize) {
            throw new IOException("Send failure for [" + text.substring(0, 32) + " ...] (MAX BUFFER SIZE " + bufferSize + "/" + bytes.length + " EXCEEDED)");
        }
        send(bytes);
    }
    
    public void receive () {
        receiver = new Thread(new Runnable() {
            public void run() {
                while ( true) {
                    try {
                        read();
                    }
                    catch (SocketException so) {
                        // socket closed?
                    }
                    catch (IOException e) {
                        e.printStackTrace();
                    }
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }, "MULTICAST-RECEIVER-" + CoreTypes.NODE_NAME);
        receiver.start();
    }
    
    private void read () throws IOException {
        // get their responses!
        byte[] buf          = new byte[bufferSize];
        DatagramPacket recv = new DatagramPacket(buf, buf.length);
        
        // this blocks until data is received
        s.receive(recv);
        
        int len = recv.getLength();
        if ( listener != null) {
            listener.onMessage(Arrays.copyOf(buf, len));
        }
    }
    
    /**
     * Get the {@link MulticastSocket} time to live (TTL)
     * @return Default TTL.
     * @throws IOException On I/O errors.
     */
    public int getTimeToLive() throws IOException {
        return s != null ? s.getTimeToLive() : 0;
    }

}

InetAddress group = InetAddress.getByName("228.5.6.7");
int port = 6789;
MulticastSocket socket = new MulticastSocket(port);
socket.joinGroup(group);

// send
String msg = “Howdy stranger”;
DatagramPacket hi = new DatagramPacket(msg.getBytes(), msg.length(), group, port);
socket.send(hi);
 
// receive
byte[] buf = new byte[1024];
DatagramPacket recv = new DatagramPacket(buf, buf.length);
socket.receive(recv);

// NetMulticaster - Sender: TestZeroSender.java
public class TestZeroSender {

    public static void main(String[] args) {
        try {
            Multicaster m = new Multicaster();
            m.joinGroup();
            m.send("Hello World");
            
            Thread.sleep(10000);
            System.out.println("Sender Shutdown.");
            m.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
// NetMulticaster - Receiver: TestZeroReceiver.java
public class TestZeroReceiver {

    public static void main(String[] args) {
        try {
            Multicaster m = new Multicaster();
            m.setListener(new MessageListener() {
                
                @Override
                public void onMessage(byte[] bytes) {
                    String s = new String(bytes);
                    System.out.println("Receiver Got: " + s + "]");
                    
                }
            });
            m.joinGroup();
            m.receive();
            
            Thread.sleep(10000);
            System.out.println("Receiver Shutdown.");
            m.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class MulticastTunnel {

    static void LOGE(String text) {
        System.err.println("[MCAST-TUNNEL] " + text);
    }

    static void LOGD(String text) {
        System.out.println("[MCAST-TUNNEL] " + text);
    }
    
    static final int DEFAULT_UDP_SERVER_PORT = 4445;
    
    // Used to receive from the remote endpoint
    private UDPServer server;

    // Used to send to the remote endpoint
    private UDPClient client;
    
    // Used to listen/bcast in the local subnet
    private NetMulticaster multicaster;
    
    // Used to prevent feedback loops
    private static final List<String> hashes = new CopyOnWriteArrayList<String>();
    
    // If true dump info to stdout
    private static boolean debug;
    
    // metric: total packets send/received
    private long packetsSent;

    // metric: total bytes sent/received
    private long bytesSent;

    // metric: uptime (ms)
    private long upTime;
    
    /**
     * The UDP server listens for packets in the local endpoint
     * 
     * @author VSilva
     *
     */
    static class UDPServer extends Thread {
        private DatagramSocket socket;
        private boolean running;
        private byte[] buf              = new byte[8192];
        NetMulticaster multicaster;
        private long bytesRecv, packetsRecv;    // metrics
        
        public UDPServer(int port , NetMulticaster caster) throws IOException {
            super("MULTICAST-TUNNEL-" + CoreTypes.NODE_NAME);
            this.socket         = new DatagramSocket(port);
            this.multicaster    = caster;
            LOGD("Started UDP Server @ port " + port);
        }
        
        @Override
        public void run() {
            running = true;
            LOGD("Running UDP server @ " + socket.getLocalAddress() + ":" + socket.getLocalPort());
            
            while (running) {
                DatagramPacket packet = new DatagramPacket(buf, buf.length);
                try {
                    // this blocks execution until a packet is received...
                    socket.receive(packet);
                    
                    final int len       = packet.getLength();
                    final byte[] data   = Arrays.copyOf(buf, len);
                    final String hash   = HASH(data);
                    
                    // must be sorted. Else won't work!
                    final int idx       = Collections.binarySearch(hashes, hash);
                    
                    if ( idx < 0 ) {
                        LOGD("UDP Server received " + len + "/" + buf.length + " bytes. Found: " + idx + " Multicasting");
                        multicaster.send(data);
                        
                        // metrics
                        packetsRecv ++;
                        bytesRecv   += len;
                        
                        addSortHashes(hash);
                    }
                    else {
                        LOGD("UDP Server received " + len + "/" + buf.length + " bytes. Hash: " + hash + " Packet already sent.");
                    }
                } 
                catch (SocketException se ) {
                    // Swallow
                }
                catch (Exception e) {
                    e.printStackTrace();
                    LOGE("UDPServer:run(): " + e.toString());
                }
            }
        }
        
        public void shutdown () throws IOException {
            running = false;
            if ( socket != null) { 
                socket.close();
            }
            packetsRecv = bytesRecv = 0;
        }
    }
    
    /**
     * Hashed packets must be sorted, else Collections.binarySearch won't work!
     * @param hash
     */
    private static void addSortHashes (final String hash) {
        hashes.add(hash);
        //Collections.sort(hashes); Cant sort CopyOnWriteArrayList - throws UnsuportedOperationException
        
        // How to sort CopyOnWriteArrayList
        // https://stackoverflow.com/questions/28805283/how-to-sort-copyonwritearraylist
        Object[] a = hashes.toArray();
        Arrays.sort(a);
        for (int i = 0; i < a.length; i++) {
            hashes.set(i, (String) a[i]);
        }   
    }
    
    /**
     * The UDP client forward packets to the remote {@link UDPServer}.
     * @author VSilva
     *
     */
    static class UDPClient {
        private DatagramSocket socket;
        private InetAddress remoteHost;
        private int remotePort;
        
        public UDPClient(String remoteHost, int remotePort) throws SocketException, UnknownHostException {
            super();
            this.remoteHost = InetAddress.getByName(remoteHost);
            this.remotePort = remotePort;
            this.socket     = new DatagramSocket();
            LOGD("Started UDP Client to remote " + remoteHost + " @ " + remotePort);
        }
        
        public void send(byte[] buf) throws IOException {
            //LOGD("UDP Client: sending " + buf.length + " bytes to " + remoteHost + ":" + remotePort);
            DatagramPacket packet = new DatagramPacket(buf, buf.length, remoteHost, remotePort);
            socket.send(packet);
        }
        
        public void shutdown () {
            if ( socket != null) {
                socket.close();
            }
        }
    }

    public MulticastTunnel() throws IOException {
    }

    /**
     * Construct a multicast tunnel.
     * @param udpServerPort The port of the local {@link UDPServer}. Default: 9876
     * @param remoteHost IP address or host name of the remote endpoint {@link UDPServer}.
     * @param remotePort Port of the remote {@link UDPServer}. Default: 9876
     * @throws IOException on I/O/Socket errors.
     */
    public MulticastTunnel(int udpServerPort , String remoteHost, int remotePort) throws IOException {
        init(udpServerPort, remoteHost, remotePort);
    }
    
    private void init (int udpServerPort , String remoteHost, int remotePort ) throws IOException {
        // 224.1.2.3 : 6789
        this.multicaster    = new NetMulticaster();
        this.server         = new UDPServer(udpServerPort, multicaster);
        this.client         = new UDPClient(remoteHost, remotePort);
        
        this.multicaster.setListener(new NetMulticaster.MessageListener() {
            
            @Override
            public void onMessage(byte[] bytes) {
                final String hash       = HASH(bytes);
                
                // must e sorted. Else won't work!
                final int idx           = Collections.binarySearch(hashes, hash);
                
                //LOGD("Multicast: got " + bytes.length + " bytes. Hash:" + hash + " found:" + idx);
                try {
                    if ( idx < 0 ) {
                        LOGD("Multicast: Tunneling " + + bytes.length + " bytes."); 
                        client.send(bytes);
                        
                        // metrics
                        packetsSent ++;
                        bytesSent   += bytes.length;
                        
                        addSortHashes(hash);
                    }
                    else {
                        //LOGD("Multicast: Packet " + hash + " already sent.");
                    }
                    
                    if ( hashes.size() > 200) {
                        LOGD("Cleaning hash list.");
                        hashes.clear();
                    }
                } catch (IOException e) {
                    LOGE(e.toString());
                }
            }
        });
        this.multicaster.open();
        this.multicaster.joinGroup();
        this.multicaster.receive();
    }
    
    public void start () throws IOException {
        if ( server == null || client == null || multicaster == null) {
            throw new IOException("Invalid constructor invoked.");
        }
        upTime  = System.currentTimeMillis();
        server.start();
    }

    public void start (int udpServerPort , String remoteHost, int remotePort ) throws IOException {
        if (isRunning()) {
            return;
        }
        upTime  = System.currentTimeMillis();
        init(udpServerPort, remoteHost, remotePort);
        server.start();
    }

    public void shutdown () throws IOException {
        LOGD("Shutting down multicast tunnel.");
        upTime = packetsSent = bytesSent = 0;
        if ( server != null) {
            server.shutdown();
        }
        if ( client != null) {
            client.shutdown();
        }
        if ( multicaster != null ) {
            multicaster.shutdown();
        }
        server      = null;
        client      = null; 
        multicaster = null;
    }

    public boolean isRunning () {
        return server != null && client != null && multicaster != null;
    }
    
    /**
     * MD5 digest tool. It should only be used to generate an instance ID. MD5 hashes are not secure.
     * @param string String to digest.
     * @return Hex encoded MD5.
     */
    static String HASH(final byte[] bytes) {
        try {
            java.security.MessageDigest md = java.security.MessageDigest.getInstance("SHA-256"); 
            byte[] digest = md.digest(bytes);
            
            StringBuffer sb = new StringBuffer();
            
            for (int i = 0; i < digest.length; ++i) {
                sb.append(Integer.toHexString((digest[i] & 0xFF) | 0x100).substring(1,3));
            }
            return sb.toString().toUpperCase();
        } 
        catch (java.security.NoSuchAlgorithmException e) {
        }
        return null;
    }

    public static void setDebug (boolean debug) {
        MulticastTunnel.debug = debug;
    }
    
    public long getPacketsSent () {
        return packetsSent;
    }
    public long getBytesSent () {
        return bytesSent;
    }

    public long getPacketsRecv () {
        return server != null ? server.packetsRecv : 0;
    }
    
    public long getBytesRecv () {
        return server != null ? server.bytesRecv : 0;
    }
    
    public long getUptime () {
        return upTime;
    }
 
}

public class TestMulticastTunnel {
  public static void main(String[] args) {
    try {
      if ( args.length < 2) {
        throw new Exception("USAGE: " + MulticastTunnel.class.getSimpleName() + " REMOTE_HOST REMOTE_PORT [DEGUG:{true/false}]");
      }
      String remoteHost   = args[0];
      int remotePort    = Integer.parseInt(args[1]);
      boolean debug       = args.length == 3 ? Boolean.parseBoolean(args[2]) : false;

      MulticastTunnel.setDebug(debug);
      MulticastTunnel tunnel = new MulticastTunnel(remotePort, remoteHost, remotePort);
      tunnel.start();
      
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
}

this.multicaster.setListener(new NetMulticaster.MessageListener() {
  
  @Override
  public void onMessage(byte[] bytes) {
    // hash packet
    final String hash     = HASH(bytes);
    
    // search for it (if found idx > 0)
    final int idx       = Collections.binarySearch(hashes, hash);
    
    try {
      if ( idx < 0 ) {
        // send via UDP
        client.send(bytes);
        // save hash int sorted hashes list 
        addSortHashes(hash);
      }
      else {
        //
      }
      // clean hash list      
      if ( hashes.size() > 200) {
        hashes.clear();
      }
    } catch (IOException e) {
      LOGE(e.toString());
    }
  }
});