NetworkHandler.java

Go to the documentation of this file.

package org.openslx.imagemaster.localrpc;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketAddress;
import java.net.SocketException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.openslx.bwlp.thrift.iface.ClientSessionData;
import org.openslx.bwlp.thrift.iface.TAuthorizationException;
import org.openslx.bwlp.thrift.iface.UserInfo;
import org.openslx.imagemaster.session.Session;
import org.openslx.imagemaster.session.SessionManager;

import com.google.gson.Gson;

public class NetworkHandler implements Runnable
{

   private static final Logger log = LogManager.getLogger( NetworkHandler.class );

   private Thread sendThread = null;
   private final Sender sender;
   private final DatagramSocket socket;
   private final Gson gson = new Gson();

   public NetworkHandler( int port, InetAddress listenAddress ) throws SocketException
   {
      socket = new DatagramSocket( port, listenAddress );
      sendThread = new Thread( sender = new Sender() );
   }

   public void shutdown()
   {
      socket.close();
   }

   private void send( SocketAddress destination, byte[] buffer )
   {
      final DatagramPacket packet;
      try {
         packet = new DatagramPacket( buffer, buffer.length, destination );
      } catch ( Exception e ) {
         log.warn( "Could not construct datagram packet for target " + destination.toString() );
         e.printStackTrace();
         return;
      }
      sender.send( packet );
   }

   @Override
   public void run()
   {
      byte readBuffer[] = new byte[ 66000 ];
      try {
         sendThread.start();
         while ( !Thread.interrupted() ) {
            DatagramPacket packet = new DatagramPacket( readBuffer, readBuffer.length );
            try {
               socket.receive( packet );
            } catch ( IOException e ) {
               log.info( "IOException on UDP socket when reading: " + e.getMessage() );
               Thread.sleep( 100 );
               continue;
            }
            if ( packet.getLength() < 2 ) {
               log.debug( "Message too short" );
               continue;
            }
            String payload = new String( readBuffer, 0, packet.getLength(), StandardCharsets.UTF_8 );
            try {
               String reply = handle( payload );
               if ( reply != null )
                  send( packet.getSocketAddress(), reply.getBytes( StandardCharsets.UTF_8 ) );
            } catch ( Throwable t ) {
               log.error( "Exception in RequestParser: " + t.toString() );
               log.error( "Payload was: " + payload );
               t.printStackTrace();
            }
         }
      } catch ( InterruptedException e ) {
         Thread.currentThread().interrupt();
      } finally {
         sendThread.interrupt();
         log.info( "UDP receiver finished." );
      }
   }

   private String handle( String payload )
   {
      try {
         JsonUser ju = gson.fromJson( payload, JsonUser.class );
         TAuthorizationException ex = ju.toException();
         if ( ex != null ) {
            SessionManager.addAuthError( ex, ju.accessCode() );
            return "Auth error";
         } else {
            UserInfo u = ju.toUser();
            if ( u == null ) {
               log.warn( "Invalid or inomplete RPC data (" + payload + ")" );
               return "Invalid or incomplete RPC data";
            }
            ClientSessionData sd = SessionManager.addSession( new Session( u, ju.timeoutSeconds() * 1000 ), ju.accessCode() );
            return "TOKEN:" + sd.authToken + " SESSIONID:" + sd.sessionId;
         }
      } catch ( Throwable t ) {
         log.error( "Exception on json decode", t );
      }
      return "Json error";
   }

   private class Sender implements Runnable
   {

      private final BlockingQueue<DatagramPacket> queue = new LinkedBlockingQueue<>( 128 );

      @Override
      public void run()
      {
         try {
            while ( !Thread.interrupted() ) {
               final DatagramPacket packet;
               packet = queue.take();
               try {
                  socket.send( packet );
               } catch ( IOException e ) {
                  log.debug( "Could not send UDP packet to " + packet.getAddress().getHostAddress().toString() );
               }
            }
         } catch ( InterruptedException e ) {
            Thread.currentThread().interrupt();
         } finally {
            log.info( "UDP sender finished." );
         }
      }

      public void send( DatagramPacket packet )
      {
         if ( queue.offer( packet ) )
            return;
         log.warn( "Could not add packet to queue: Full" );
      }

   }

}