packetcable-driver/src/main/java/org/pcmm/rcd/impl/AbstractPCMMServer.java

   1 /*
   2  * Copyright (c) 2014, 2015 Cable Television Laboratories, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.pcmm.rcd.impl;
  10
  11 import java.util.concurrent.atomic.AtomicBoolean;
  12 import org.pcmm.PCMMConstants;
  13 import org.pcmm.PCMMProperties;
  14 import org.pcmm.concurrent.IWorkerPool;
  15 import org.pcmm.concurrent.impl.WorkerPool;
  16 import org.pcmm.rcd.IPCMMServer;
  17 import org.pcmm.state.IState;
  18 import org.slf4j.Logger;
  19 import org.slf4j.LoggerFactory;
  20
  21 import java.io.IOException;
  22 import java.net.ServerSocket;
  23 import java.net.Socket;
  24 import java.util.concurrent.ExecutorService;
  25 import java.util.concurrent.Executors;
  26
  27 // import org.junit.Assert;
  28
  29 /*
  30  * (non-Javadoc)
  31  *
  32  * @see pcmm.rcd.IPCMMServer
  33  */
  34 public abstract class AbstractPCMMServer implements IPCMMServer {
  35
  36     private static final Logger logger = LoggerFactory.getLogger(AbstractPCMMServer.class);
  37
  38     /*
  39      * A ServerSocket to accept messages ( OPN requests)
  40      */
  41     private transient ServerSocket serverSocket;
  42
  43     private volatile boolean keepAlive;
  44
  45     /*
  46      * This object is used to signal when the server is ready to accept connections.
  47      * It is used in signaling and therefore must be final.
  48      */
  49     private final AtomicBoolean isReadyFlag = new AtomicBoolean(false);
  50
  51     /**
  52      * The port number on which to start the server.
  53      */
  54     private final int port;
  55
  56     IWorkerPool pool;
  57
  58     /**
  59      * The thread pool executor
  60      */
  61     private final ExecutorService executorService;
  62
  63     /**
  64      * Constructor to use the port number contained within the PCMMProperties static object
  65      */
  66     protected AbstractPCMMServer() {
  67         this(PCMMProperties.get(PCMMConstants.PCMM_PORT, Integer.class));
  68     }
  69
  70     /**
  71      * Constructor for starting the server to a pre-defined port number. When 0 is used, the server socket will
  72      * assign one for you. To determine which port is being used, call getPort() after startServer() is called.
  73      * @param port - the port number on which to start the server
  74      */
  75     protected AbstractPCMMServer(int port) {
  76         // XXX - Assert.assertTrue(port >= 0 && port <= 65535);
  77         this.port = port;
  78         keepAlive = true;
  79         int poolSize = PCMMProperties.get(PCMMConstants.PS_POOL_SIZE, Integer.class);
  80         pool = new WorkerPool(poolSize);
  81         executorService = Executors.newSingleThreadExecutor();
  82     }
  83
  84     /*
  85      * (non-Javadoc)
  86      *
  87      * @see pcmm.rcd.IPCMMServer#startServer()
  88      */
  89     public void startServer() throws IOException {
  90         if (serverSocket != null)
  91             return;
  92
  93         serverSocket = new ServerSocket(port);
  94         logger.info("Server started and listening on port :" + port);
  95
  96         // execute this in a single thread executor
  97         executorService.execute(new Runnable() {
  98             public void run() {
  99
 100                 synchronized (isReadyFlag) {
 101                     isReadyFlag.set(true);
 102                     isReadyFlag.notify();
 103                 }
 104
 105                 while (keepAlive) {
 106                     try {
 107                         Socket socket = serverSocket.accept();
 108                         logger.info("Accepted a new connection from :" + socket.getInetAddress().getHostAddress() + ":" + socket.getPort());
 109                         if (keepAlive) {
 110                             pool.schedule(getPCMMClientHandler(socket));
 111                             logger.info("Handler attached to : " + socket.getInetAddress().getHostAddress() + ":" + socket.getPort());
 112                         } else {
 113                             logger.info("connection to be closed : " + socket.getInetAddress().getHostAddress() + ":" + socket.getPort());
 114                             socket.close();
 115                         }
 116                     } catch (IOException e) {
 117                         logger.error(e.getMessage());
 118                     }
 119                 }
 120
 121                 synchronized (isReadyFlag) {
 122                     isReadyFlag.set(false);
 123                 }
 124                 stopServer();
 125
 126             }
 127         });
 128
 129         synchronized (isReadyFlag) {
 130             // block until the server thread is ready to go
 131             while (!isReadyFlag.get()) {
 132                 try {
 133                     isReadyFlag.wait();
 134                 } catch (InterruptedException e) {
 135                     Thread.currentThread().interrupt();
 136                 }
 137             }
 138         }
 139     }
 140
 141     public boolean isReady() {
 142         return isReadyFlag.get();
 143     }
 144
 145     /**
 146      * This client is used to handle requests from within the Application
 147      * Manager
 148      *
 149      * @param socket - the connection to the PCMM server
 150      * @return client handler
 151      */
 152     protected abstract IPCMMClientHandler getPCMMClientHandler(Socket socket) throws IOException;
 153
 154     @Override
 155     public void stopServer() {
 156         // set to stop
 157         keepAlive = false;
 158         executorService.shutdownNow();
 159         try {
 160             if (serverSocket != null) {
 161                 serverSocket.close();
 162             }
 163         } catch (Exception e) {
 164             logger.error(e.getMessage());
 165         }
 166         pool.killAll();
 167     }
 168
 169     @Override
 170     public void recordState() {
 171         // TODO - implement me
 172     }
 173
 174     @Override
 175     public IState getRecoredState() {
 176         return null;
 177     }
 178
 179     /**
 180      * @return the port
 181      */
 182     public int getPort() {
 183         if (serverSocket != null && serverSocket.isBound()) return serverSocket.getLocalPort();
 184         else return this.port;
 185     }
 186
 187 }