path: root/src/main/java/org/openslx/taskmanager/tasks/DispatchRelay.java

package org.openslx.taskmanager.tasks;

import java.net.Socket;
import java.security.cert.X509Certificate;

import javax.net.SocketFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLSocket;
import javax.net.ssl.SSLSocketFactory;
import javax.net.ssl.TrustManager;
import javax.net.ssl.X509TrustManager;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetSocketAddress;
import org.openslx.taskmanager.api.AbstractTask;

import com.google.gson.annotations.Expose;

public class DispatchRelay extends AbstractTask {

	protected Output status = new Output();
	
	@Expose
	public String[] hosts;
	@Expose
	public int[] ports;
	@Expose
	public String[] descs;
	
	@Override
	protected boolean execute() 
	{
		Socket[] sockets;
		Thread[] relayThreads;
		try {
			sockets = createSockets( getSocketFactory() );
			status.addMessage( "Sockets connected." );
		} catch ( Exception e ) {
			status.addMessage( "Failed to connect sockets. " + e.getMessage() );
			return false;
		}
		
		try {
			relayThreads = createRelayThreads( sockets );
			status.addMessage( "Relays created." );
		} catch ( IOException iox ) {
			status.addMessage( "Failed to create relays. " + iox.getMessage() );
			return false;
		}
		
		for ( Thread t : relayThreads )
			t.start();
		
		try {
			for ( Thread t : relayThreads )
				t.join();
		} catch ( InterruptedException inx ) {} finally {
			try {
				for ( Socket s : sockets ) {
					if ( !s.isClosed() )
						s.close();
				}
				status.addMessage( "Sockets closed properly." );
			} catch ( IOException iox ) {
				status.addMessage( "Failed to close Sockets. " + iox.getMessage() );	
			}	
		}
		return true;
	}

	@Override
	protected boolean initTask() 
	{
		this.setStatusObject( this.status );
		
		if ( hosts.length != 2 || ports.length != 2 ) {
			this.status.addMessage( "Invalid host/port list." );
			return false;
		}
		
		this.status.addMessage( "Initiated relay task: " + this.getClass().getName() );
		this.status.addMessage( this.descs[0] + " is " + this.hosts[0] + ":" + this.ports[0] );
		this.status.addMessage( this.descs[1] + " is " + this.hosts[1] + ":" + this.ports[1] );
		return true;
	}
	
	protected Thread[] createRelayThreads ( Socket[] sockets) throws IOException
	{
		Thread[] t = new Thread[2];
		Socket[][] s = { sockets, { sockets[1], sockets[0] }};
		
		for ( int i = 0; i < 2; i++ ) {
			t[i] = new Thread( new Relay( s[i], status, descs[i] ));
		}
		return t;
	}
	
	protected SSLSocketFactory getSocketFactory () throws Exception {
		status.addMessage( "Using TLS/SSL encryption." );
		return trustAll().getSocketFactory();
	}
	
	protected SSLContext trustAll () throws Exception {
		TrustManager[] trustAllMan = new TrustManager[] { new X509TrustManager() {
				public java.security.cert.X509Certificate[] getAcceptedIssuers() {
					return null;
				}
				public void checkClientTrusted( X509Certificate[] certs, String authType ) {}
				public void checkServerTrusted( X509Certificate[] certs, String authType ) {}
			}
		};
		SSLContext ctx = SSLContext.getInstance( "SSL" );
		ctx.init( null, trustAllMan, new java.security.SecureRandom() );
		return ctx;
	}
	
	protected SSLSocket[] createSockets ( SocketFactory sf ) throws IOException 
	{		
		SSLSocket[] s = new SSLSocket[2];
		
		for ( int i = 0; i < 2; i++ ) {
			InetSocketAddress addr = new InetSocketAddress( hosts[i], ports[i] );
			s[i] = (SSLSocket) sf.createSocket();
			this.status.addMessage( "trying to connect socket to " + addr.toString() );
			s[i].connect( addr, 1200 );
			this.status.addMessage( "connected." );
		}
		return s;
	}
	
	public static class Output 
	{		
		protected String messages = null;
		
		public void addMessage( String str )
		{
			if ( messages == null ) 
			{
				messages = str;
			} else {
				messages += "\n" + str;
			}
		}
	}
	
	private class Relay implements Runnable {
	
		private InputStream in;
		private OutputStream out;
		
		private Output status;
		public final String desc;
		
		private byte[] buffer = new byte[16384];
		
		public Relay ( Socket[] s, Output status, String desc ) throws IOException 
		{		
			in = s[0].getInputStream();
			out = s[1].getOutputStream();
			this.status = status;
			this.desc = desc;
		}
		
		private void relay() throws IOException, InterruptedException 
		{		
			int readBytes = in.read( buffer );
			out.write( buffer, 0, readBytes );
		}
		
		public void close() 
		{
			try {
				if ( this.in != null )
					this.in.close();
				if ( this.out != null )
					this.out.close();
				this.status.addMessage( desc + ": input/output streams closed properly." );
			} catch ( IOException iox ) {
				this.status.addMessage( desc + ": failed to close input/ouput streams. " + iox.getMessage() );
			}	
		}
		
		@Override
		public void run() 
		{
			boolean first = true;
			while ( true ) {
				try {
					relay();
					if ( first ) {
						this.status.addMessage( desc + ": relay operating." );
						first = false;
					}			
				} catch ( Exception x ) {
					this.status.addMessage( desc + ": close relay. " );
					this.close();
					return;
				}
			}	
		}
	}
}