/*

  Copyright 2007 Alexey Akhunov

  Licensed under the Apache License, Version 2.0 (the "License");

  you may not use this file except in compliance with the License.

  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software

  distributed under the License is distributed on an "AS IS" BASIS,

  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

  See the License for the specific language governing permissions and

  limitations under the License.

  */

 package diy.middleware.sockets;

 import java.io.BufferedInputStream;

 import java.io.BufferedOutputStream;

 import java.io.DataInputStream;

 import java.io.DataOutputStream;

 import java.io.EOFException;

 import java.io.OutputStream;

 import java.net.Socket;

 import java.util.HashMap;

 import java.util.LinkedList;

 import java.util.Map;

 import java.util.concurrent.BlockingQueue;

 import java.util.concurrent.LinkedBlockingQueue;

 import java.util.concurrent.locks.ReadWriteLock;

 import java.util.concurrent.locks.ReentrantReadWriteLock;

 import diy.middleware.Frame;

 import diy.middleware.Monitor;

 import diy.middleware.SessionCommand;

 public class SocketTransport implements Runnable {

         private class SocketQueueScanner implements Runnable {

                 public void run() {

                         while (true) {

                                 try {

                                         final SocketConnection incomingConnection = socketQueue

                                                         .take();

                                         final String sessionId = incomingConnection.getSessionId();

                                         OutputStream output = incomingConnection.getSocket()

                                                         .getOutputStream();

                                         final DataOutputStream dataOutput = new DataOutputStream(

                                                         new BufferedOutputStream(output));

                                         final BlockingQueue<Frame> queue = new LinkedBlockingQueue<Frame>();

                                         final InternalSession internalSession = new InternalSession();

                                         internalSession.socket = incomingConnection.getSocket();

                                         Thread outputThread = new Thread(new Runnable() {

                                                 public void run() {

                                                         try {

                                                                 while (true) {

                                                                         Frame outputFrame = queue.take();

                                                                         monitor.count(-1, "Output|queue");

                                                                         dataOutput.writeUTF(outputFrame

                                                                                         .getContent());

                                                                         dataOutput.flush();

                                                                 }

                                                         } catch (InterruptedException ie) {

                                                         } catch (Exception exception) {

                                                                 destroySession(sessionId, exception, true);

                                                         }

                                                 }

                                         });

                                         internalSession.outputThread = outputThread;

                                         try {

                                                 internalSessionsLock.writeLock().lock();

                                                 internalSessions.put(sessionId,internalSession);

                                         } finally {

                                                 internalSessionsLock.writeLock().unlock();

                                         }

                                         outputThread.start();

                                         SessionCommand command = new SessionCommand(

                                                         incomingConnection.getSessionId(),

                                                         SessionCommand.Command.QUEUE);

                                         command.setQueue(queue);

                                         outQueue.put(command);

                                 } catch (InterruptedException ie) {

                                         return;

                                 } catch (Exception e) {

                                         e.printStackTrace();

                                 }

                         }

                 }

         }

         private BlockingQueue<SocketConnection> socketQueue;

         private static class InternalSession {

                 private Thread inputThread;

                 private Thread outputThread;

                 private Socket socket;

         }

         private Map<String, InternalSession> internalSessions = new HashMap<String, InternalSession>();

         private ReadWriteLock internalSessionsLock = new ReentrantReadWriteLock();

         private BlockingQueue<SessionCommand> inQueue;

         private BlockingQueue<SessionCommand> outQueue;

         private Thread transportThread;

         private Thread scannerThread;

         private Monitor monitor;

         public void init() {

                 transportThread = new Thread(this);

                 transportThread.start();

                 scannerThread = new Thread(new SocketQueueScanner());

                 scannerThread.start();

         }

         public void run() {

                 while (true) {

                         try {

                                 final SessionCommand sessionCommand = inQueue.take();

                                 switch (sessionCommand.getCommand()) {

                                 case QUEUE: {

                                         final String sessionId = sessionCommand.getSessionId();

                                         final InternalSession internalSession;

                                         try {

                                                 internalSessionsLock.readLock().lock();

                                                 internalSession = internalSessions.get(sessionId);

                                         } finally {

                                                 internalSessionsLock.readLock().unlock();

                                         }

                                         if (internalSession == null) {

                                                 // TODO: throw exception

                                                 break;

                                         }

                                         final BlockingQueue<Frame> queue = sessionCommand

                                                         .getQueue();

                                         final DataInputStream dataInput = new DataInputStream(

                                                         new BufferedInputStream(internalSession.socket

                                                                         .getInputStream()));

                                         Thread inputThread = new Thread(new Runnable() {

                                                 public void run() {

                                                         try {

                                                                 while (true) {

                                                                         try {

                                                                                 String incomingMessage = dataInput

                                                                                                 .readUTF();

                                                                                 queue.put(new Frame(sessionId,

                                                                                                 incomingMessage));

                                                                                 monitor.count(1, "Distribution|queue");

                                                                         } catch (EOFException eof) {

                                                                                 // Socket closed by peer

                                                                                 internalSession.outputThread

                                                                                                 .interrupt();

                                                                                 try {

                                                                                         internalSession.socket.close();

                                                                                 } catch (Exception se) {

                                                                                 }

                                                                                 SessionCommand command = new SessionCommand(

                                                                                                 sessionId,

                                                                                                 SessionCommand.Command.CLOSE);

                                                                                 outQueue.put(command);

                                                                                 break;

                                                                         }

                                                                 }

                                                         } catch (InterruptedException ie) {

                                                         } catch (Exception exception) {

                                                                 destroySession(sessionId, exception, true);

                                                         }

                                                 }

                                         });

                                         internalSession.inputThread = inputThread;

                                         inputThread.start();

                                         break;

                                 }

                                 case CLOSE: {

                                         final String sessionId = sessionCommand.getSessionId();

                                         destroySession(sessionId, null, false);

                                         break;

                                 }

                                 }

                         } catch (InterruptedException ie) {

                                 // Gracefull exit

                                 break;

                         } catch (Exception e) {

                                 e.printStackTrace();

                         }

                 }

         }

         public void setInQueue(BlockingQueue<SessionCommand> inQueue) {

                 this.inQueue = inQueue;

         }

         public void setOutQueue(BlockingQueue<SessionCommand> outQueue) {

                 this.outQueue = outQueue;

         }

         public void setMonitor(Monitor monitor) {

                 this.monitor = monitor;

         }

         public void setSocketQueue(BlockingQueue<SocketConnection> socketQueue) {

                 this.socketQueue = socketQueue;

         }

         public void destroy() {

                 scannerThread.interrupt();

                 transportThread.interrupt();

                 try {

                         internalSessionsLock.writeLock().lock();

                         LinkedList<String> ids = new LinkedList<String>(internalSessions.keySet());

                         for (String sessionId: ids) {

                                 destroySession(sessionId, null, false);

                         }

                 } finally {

                         internalSessionsLock.writeLock().unlock();

                 }

         }

         private void destroySession(String sessionId, Exception exception, boolean send) {

                 InternalSession session;

                 try {

                         internalSessionsLock.writeLock().lock();

                         session = internalSessions.remove(sessionId);

                 } finally {

                         internalSessionsLock.writeLock().unlock();

                 }

                 if (session == null) {

                         // Already removed

                         return;

                 }

                 try {

                         session.socket.close();

                 } catch (Exception e) {};

                 if (session.inputThread != null && Thread.currentThread() != session.inputThread) {

                         session.inputThread.interrupt();

                 }

                 if (Thread.currentThread() != session.outputThread) {

                         session.outputThread.interrupt();

                 }

                 if (send) {

                         SessionCommand command = new SessionCommand(

                                 sessionId, SessionCommand.Command.CLOSE);

                         command.setException(exception);

                         // If we cannot quickly send close command, then nevermind

                         outQueue.offer(command);

                 }

         }

 }