packetcable-driver/src/main/java/org/umu/cops/prpep/COPSPepConnection.java

   1 /*
   2  * Copyright (c) 2004 University of Murcia.  All rights reserved.
   3  * --------------------------------------------------------------
   4  * For more information, please see <http://www.umu.euro6ix.org/>.
   5  */
   6
   7 package org.umu.cops.prpep;
   8
   9 import org.slf4j.Logger;
  10 import org.slf4j.LoggerFactory;
  11 import org.umu.cops.COPSConnection;
  12 import org.umu.cops.stack.*;
  13 import org.umu.cops.stack.COPSDecision.Command;
  14 import org.umu.cops.stack.COPSDecision.DecisionFlag;
  15
  16 import javax.annotation.concurrent.ThreadSafe;
  17 import java.io.IOException;
  18 import java.net.Socket;
  19 import java.util.Date;
  20 import java.util.Map;
  21 import java.util.Set;
  22 import java.util.concurrent.ConcurrentHashMap;
  23
  24 /**
  25  * COPSPepConnection represents a PEP-PDP Connection Manager.
  26  * Responsible for processing messages received from PDP.
  27  */
  28 @ThreadSafe
  29 public class COPSPepConnection extends COPSConnection {
  30
  31     private final static Logger logger = LoggerFactory.getLogger(COPSPepConnection.class);
  32
  33     /** Time to wait responses (milliseconds), default is 10 seconds */
  34     protected transient int _responseTime;
  35
  36     /** COPS Client-type */
  37     protected final short _clientType;
  38
  39     /**
  40      Maps a COPS Client Handle to a Request State Manager
  41      */
  42     protected final Map<COPSHandle, COPSPepReqStateMan> _managerMap;
  43
  44     /**
  45      * Creates a new PEP connection
  46      * @param clientType    PEP's client-type
  47      * @param sock          Socket connected to PDP
  48      */
  49     public COPSPepConnection(final short clientType, final Socket sock) {
  50         super(sock, (short)0, (short)0);
  51         _clientType = clientType;
  52         _responseTime = 10000;
  53         _managerMap = new ConcurrentHashMap<>();
  54     }
  55
  56     /**
  57      * Message-processing loop
  58      */
  59     public void run () {
  60         Date lastSendKa = new Date();
  61         Date lastSendAcc = new Date();
  62         Date lastRecKa = new Date();
  63         try {
  64             while (!_sock.isClosed()) {
  65                 if (_sock.getInputStream().available() != 0) {
  66                     processMessage(_sock);
  67                     lastRecKa = new Date();
  68                 }
  69
  70                 // Keep Alive
  71                 if (_kaTimer > 0) {
  72                     // Timeout at PDP
  73                     int _startTime = (int) (lastRecKa.getTime());
  74                     int cTime = (int) (new Date().getTime());
  75
  76                     if ((cTime - _startTime) > _kaTimer*1000) {
  77                         _sock.close();
  78                         // Notify all Request State Managers
  79                         notifyNoKAAllReqStateMan();
  80                     }
  81
  82                     // Send to PEP
  83                     _startTime = (int) (lastSendKa.getTime());
  84                     cTime = (int) (new Date().getTime());
  85
  86                     if ((cTime - _startTime) > ((_kaTimer*3/4) * 1000)) {
  87                         final COPSKAMsg msg = new COPSKAMsg(null);
  88                         COPSTransceiver.sendMsg(msg, _sock);
  89                         lastSendKa = new Date();
  90                     }
  91                 }
  92
  93                 // Accounting
  94                 if (_acctTimer > 0) {
  95                     int _startTime = (int) (lastSendAcc.getTime());
  96                     int cTime = (int) (new Date().getTime());
  97
  98                     if ((cTime - _startTime) > ((_acctTimer*3/4)*1000)) {
  99                         // Notify all Request State Managers
 100                         notifyAcctAllReqStateMan();
 101                         lastSendAcc = new Date();
 102                     }
 103                 }
 104
 105                 try {
 106                     Thread.sleep(500);
 107                 } catch (Exception e) {
 108                     logger.error("Exception thrown while sleeping", e);
 109                 }
 110             }
 111         } catch (Exception e) {
 112             logger.error("Error while processing socket messages", e);
 113         }
 114
 115         // connection closed by server
 116         // COPSDebug.out(getClass().getName(),"Connection closed by server");
 117         try {
 118             _sock.close();
 119         } catch (IOException e) {
 120             logger.error("Error closing socket", e);
 121         }
 122
 123         // Notify all Request State Managers
 124         try {
 125             notifyCloseAllReqStateMan();
 126         } catch (COPSException e) {
 127             logger.error("Error closing state managers");
 128         }
 129     }
 130
 131     /**
 132      * Gets a COPS message from the socket and processes it
 133      * @param conn  Socket connected to the PDP
 134      * @throws COPSException
 135      * @throws IOException
 136      */
 137     protected void processMessage(final Socket conn) throws COPSException, IOException {
 138         final COPSMsg msg = COPSTransceiver.receiveMsg(conn);
 139
 140         switch (msg.getHeader().getOpCode()) {
 141             case CC:
 142                 handleClientCloseMsg(conn, (COPSClientCloseMsg)msg);
 143                 break;
 144             case DEC:
 145                 handleDecisionMsg((COPSDecisionMsg)msg);
 146                 break;
 147             case SSQ:
 148                 handleSyncStateReqMsg((COPSSyncStateMsg)msg);
 149                 break;
 150             case KA:
 151                 handleKeepAliveMsg((COPSKAMsg)msg);
 152                 break;
 153             default:
 154                 throw new COPSPepException("Message not expected (" + msg.getHeader().getOpCode() + ").");
 155         }
 156     }
 157
 158     /**
 159      * Handle Keep Alive Message
 160      * @param    cMsg                a  COPSKAMsg
 161      */
 162     private void handleKeepAliveMsg(final COPSKAMsg cMsg) {
 163         logger.info("Get KAlive Msg");
 164         try {
 165             // Support
 166             if (cMsg.getIntegrity() != null) {
 167                 logger.warn("Unsupported objects (Integrity)");
 168             }
 169
 170             // should we do anything else?? ....
 171
 172         } catch (Exception unae) {
 173             logger.error("Unexpected exception while writing COPS data", unae);
 174         }
 175     }
 176
 177     /**
 178      * Method handleDecisionMsg
 179      * @param    dMsg                 a  COPSDecisionMsg
 180      */
 181     protected void handleDecisionMsg(final COPSDecisionMsg dMsg) throws COPSException {
 182         final COPSHandle handle = dMsg.getClientHandle();
 183         final Map<COPSContext, Set<COPSDecision>> decisions = dMsg.getDecisions();
 184
 185         for (final Set<COPSDecision> copsDecisions: decisions.values()) {
 186             for (final COPSDecision decision : copsDecisions) {
 187                 // Get the associated manager
 188                 final COPSPepReqStateMan manager = _managerMap.get(handle);
 189                 if (manager == null) {
 190                     logger.warn("Unable to find state manager with key - " + handle);
 191                     return;
 192                 }
 193
 194                 // Check message type
 195                 // TODO FIXME - Use of manager object could result in a NPE
 196                 if (decision.getFlag().equals(DecisionFlag.REQSTATE)) {
 197                     if (decision.getCommand().equals(Command.REMOVE))
 198                         // Delete Request State
 199                         manager.processDeleteRequestState(dMsg);
 200                     else
 201                         // Open new Request State
 202                         handleOpenNewRequestStateMsg(handle);
 203                 } else
 204                     // Decision
 205                     manager.processDecision(dMsg);
 206             }
 207         }
 208     }
 209
 210
 211     /**
 212      * Method handleOpenNewRequestStateMsg
 213      * @param    handle              a  COPSHandle
 214      */
 215     private void handleOpenNewRequestStateMsg(final COPSHandle handle) throws COPSPepException {
 216         final COPSPepReqStateMan manager = _managerMap.get(handle);
 217         if (manager == null)
 218             logger.warn("Unable to find state manager with key - " + handle.getId().str());
 219         else
 220             manager.processOpenNewRequestState();
 221     }
 222
 223     /**
 224      * Method handleSyncStateReqMsg
 225      * @param    cMsg                a  COPSSyncStateMsg
 226      */
 227     private void handleSyncStateReqMsg(final COPSSyncStateMsg cMsg) throws COPSException {
 228         if (cMsg.getIntegrity() != null) {
 229             logger.warn("Unsupported objects (Integrity)");
 230         }
 231
 232         final COPSPepReqStateMan manager = _managerMap.get(cMsg.getClientHandle());
 233         if (manager == null)
 234             logger.warn("Unable to find state manager with key - " + cMsg.getClientHandle().getId().str());
 235         else
 236             manager.processSyncStateRequest(cMsg);
 237     }
 238
 239     /**
 240      * Method createRequestState
 241      * @param    clientHandle             a  String
 242      * @param    process                  a  COPSPepDataProcess
 243      * @return   a COPSPepmanager
 244      * @throws   COPSException
 245      * @throws   COPSPepException
 246      */
 247     public COPSPepReqStateMan addRequestState(final COPSHandle clientHandle, final COPSPepDataProcess process)
 248             throws COPSException {
 249         final COPSPepReqStateMan manager = new COPSPepReqStateMan(_clientType, clientHandle, process, _sock);
 250         if (_managerMap.get(clientHandle) != null)
 251             throw new COPSPepException("Duplicate Handle, rejecting " + clientHandle);
 252
 253         _managerMap.put(clientHandle, manager);
 254         logger.info("Added state manager with key - " + clientHandle);
 255         manager.initRequestState();
 256         return manager;
 257     }
 258
 259     /**
 260      * Method deleteRequestState
 261      * @param    manager             a  COPSPepReqStateMan
 262      * @throws   COPSException
 263      */
 264     public void deleteRequestState(COPSPepReqStateMan manager) throws COPSException {
 265         manager.finalizeRequestState();
 266     }
 267
 268     private void notifyCloseAllReqStateMan() throws COPSException {
 269         for (final COPSPepReqStateMan man: _managerMap.values()) {
 270             man.processClosedConnection(_error);
 271         }
 272     }
 273
 274     private void notifyNoKAAllReqStateMan() throws COPSException {
 275         for (final COPSPepReqStateMan man: _managerMap.values()) {
 276             man.processNoKAConnection();
 277         }
 278     }
 279
 280     private void notifyAcctAllReqStateMan() throws COPSException {
 281         for (final COPSPepReqStateMan man: _managerMap.values()) {
 282             man.processAcctReport();
 283         }
 284     }
 285
 286 }
 287