/*

         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.client;

 import java.io.IOException;

 import java.net.UnknownHostException;

 import java.rmi.server.UID;

 import java.util.HashMap;

 import java.util.Map;

 import java.util.concurrent.BlockingQueue;

 import java.util.concurrent.CountDownLatch;

 import java.util.concurrent.LinkedBlockingQueue;

 import java.util.concurrent.TimeUnit;

 import java.util.concurrent.locks.ReadWriteLock;

 import java.util.concurrent.locks.ReentrantReadWriteLock;

 import diy.middleware.Connector;

 import diy.middleware.Frame;

 import diy.middleware.SessionCommand;

 public class Client implements Runnable {

         private BlockingQueue<SessionCommand> inQueue;

         private BlockingQueue<SessionCommand> outQueue;

         private Connector connector;

         private Map<String,Session> sessionMap = new HashMap<String,Session>();

         private ReadWriteLock sessionMapLock = new ReentrantReadWriteLock();

         public void setInQueue(BlockingQueue<SessionCommand> inQueue) {

                 this.inQueue = inQueue;

         }

         public void setOutQueue(BlockingQueue<SessionCommand> outQueue) {

                 this.outQueue = outQueue;

         }

         public Session createSession(String host, int port, BlockingQueue<Frame> inMessageQueue, long timeout, TimeUnit unit) throws UnknownHostException, IOException, InterruptedException {

                 String sessionId = new UID().toString();

                 // Create a latch

                 CountDownLatch latch = new CountDownLatch(1);

                 Session session = new Session(sessionId, inMessageQueue, latch);

                 try {

                         sessionMapLock.writeLock().lock();

                         sessionMap.put(sessionId, session);

                 } finally {

                         sessionMapLock.writeLock().unlock();

                 }

                 connector.connect(host, port, sessionId);

                 if (!latch.await(timeout, unit)) {

                         return null;

                 }

                 // Send QUEUE command to the transport

                 SessionCommand command = new SessionCommand(sessionId, SessionCommand.Command.QUEUE);

                 command.setQueue(inFrameQueue);

                 outQueue.put(command);

                 return session;

         }

         public void setConnector(Connector connector) {

                 this.connector = connector;

         }

         private Thread thread;

         private int threads;

         private Thread[] dispatchThreads;

         private LinkedBlockingQueue<Frame> inFrameQueue = new LinkedBlockingQueue<Frame>();

         public void init() {

                 dispatchThreads = new Thread[threads];

                 for (int i=0;i<threads;i++) {

                         dispatchThreads[i] = new Thread(new Runnable() {

                                 public void run() {

                                         try {

                                         while (true) {

                                                 Frame frame = inFrameQueue.take();

                                                 Session session;

                                                 try {

                                                         sessionMapLock.readLock().lock();

                                                         session = sessionMap.get(frame.getSessionId());

                                                 } finally {

                                                         sessionMapLock.readLock().unlock();

                                                 }

                                                 if (frame.getContent().startsWith("REC") || frame.getContent().equals("STOP")) {

                                                         // subscription acknowledgement arrived

                                                         session.getLatch().countDown();

                                                 } else if (frame.getContent().startsWith("M")){

                                                         session.getInQueue().put(new Frame(frame.getSessionId(),frame.getContent().substring(1)));

                                                 }

                                         }

                                         } catch (InterruptedException e) {

                                         }

                                 }

                         });

                         dispatchThreads[i].start();

                 }

                 thread = new Thread(this);

                 thread.start();

         }

         public void run() {

                 try {

                         while (true) {

                                 SessionCommand command = inQueue.take();

                                 Session session;

                                 try {

                                         sessionMapLock.readLock().lock();

                                         session = sessionMap.get(command.getSessionId());

                                 } finally {

                                         sessionMapLock.readLock().unlock();

                                 }

                                 switch (command.getCommand()) {

                                 case QUEUE:

                                         session.setOutQueue(command.getQueue());

                                         session.getLatch().countDown();

                                         break;

                                 case CLOSE:

                                         command.getException().printStackTrace();

                                         try {

                                                 sessionMapLock.writeLock().lock();

                                                 sessionMap.remove(session.getSessionId());

                                         } finally {

                                                 sessionMapLock.writeLock().unlock();

                                         }

                                 }

                         }

                 } catch (InterruptedException e) {

                 }

         }

         public void setThreads(int threads) {

                 this.threads = threads;

         }

         public void closeSession(Session session) throws InterruptedException {

                 try {

                         sessionMapLock.writeLock().lock();

                         sessionMap.remove(session.getSessionId());

                 } finally {

                         sessionMapLock.writeLock().unlock();

                 }

                 SessionCommand command = new SessionCommand(session.getSessionId(), SessionCommand.Command.CLOSE);

                 outQueue.put(command);

         }

         public void destroy() {

                 thread.interrupt();

                 for (int i=0;i<threads;i++) {

                         dispatchThreads[i].interrupt();

                 }

         }

 }