packetcable-driver/src/main/java/org/pcmm/PCMMPdpConnection.java

   1 /*
   2  @header@
   3  */
   4
   5 package org.pcmm;
   6
   7 import org.slf4j.Logger;
   8 import org.slf4j.LoggerFactory;
   9 import org.umu.cops.prpdp.COPSPdpException;
  10 import org.umu.cops.stack.*;
  11 import org.umu.cops.stack.COPSHeader.ClientType;
  12 import org.umu.cops.stack.COPSHeader.OPCode;
  13
  14 import java.io.IOException;
  15 import java.net.Socket;
  16 import java.util.Date;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import java.util.concurrent.ConcurrentHashMap;
  20
  21 /**
  22  * Class for managing an provisioning connection at the PDP side.
  23  */
  24 public class PCMMPdpConnection implements Runnable {
  25
  26
  27     public final static Logger logger = LoggerFactory.getLogger(PCMMPdpConnection.class);
  28
  29     /**
  30     Socket connected to PEP
  31      */
  32     private Socket _sock;
  33
  34     /**
  35     PEP identifier
  36     */
  37     private COPSPepId _pepId;
  38
  39     /**
  40      *  Time of the latest keep-alive received
  41      */
  42     protected Date _lastRecKa;
  43
  44     /**
  45    Maps a Client Handle to a Handler
  46      */
  47     protected Map<String, PCMMPdpReqStateMan> _managerMap;
  48
  49     /**
  50      *  PDP policy data processor class
  51      */
  52     protected PCMMPdpDataProcess _process;
  53
  54     /**
  55    Accounting timer value (secs)
  56      */
  57     protected short _acctTimer;
  58
  59     /**
  60    Keep-alive timer value (secs)
  61      */
  62     protected short _kaTimer;
  63
  64     /**
  65    COPS error returned by PEP
  66      */
  67     protected COPSError _error;
  68
  69     /**
  70      * Creates a new PDP connection
  71      *
  72      * @param pepId PEP-ID of the connected PEP
  73      * @param sock Socket connected to PEP
  74      * @param process Object for processing policy data
  75      */
  76     public PCMMPdpConnection(COPSPepId pepId, Socket sock, PCMMPdpDataProcess process) {
  77         _sock = sock;
  78         _pepId = pepId;
  79         _managerMap = new ConcurrentHashMap<>();
  80         _kaTimer = 120;
  81         _process = process;
  82     }
  83
  84     /**
  85      * Sets the keep-alive timer value
  86      * @param kaTimer Keep-alive timer value (secs)
  87      */
  88     public void setKaTimer(final short kaTimer) {
  89         _kaTimer = kaTimer;
  90     }
  91
  92     /**
  93      * Sets the accounting timer value
  94      * @param acctTimer Accounting timer value (secs)
  95      */
  96     public void setAccTimer(final short acctTimer) {
  97         _acctTimer = acctTimer;
  98     }
  99
 100     /**
 101      * Gets the accounting timer value
 102      * @return Accounting timer value (secs)
 103      */
 104     public short getAcctTimer() {
 105         return _acctTimer;
 106     }
 107
 108     /**
 109      * Gets the handle map
 110      * @return   A <tt>Hashtable</tt> holding the handle map
 111      */
 112     public Map<String, PCMMPdpReqStateMan> getReqStateMans() {
 113         // Defensive copy
 114         return new HashMap<>(_managerMap);
 115     }
 116
 117     /**
 118      * Checks whether the socket to the PEP is closed or not
 119      * @return   <tt>true</tt> if closed, <tt>false</tt> otherwise
 120      */
 121     public boolean isClosed() {
 122         return _sock.isClosed();
 123     }
 124
 125     /**
 126      * Closes the socket to the PEP
 127      * @throws IOException
 128      */
 129     protected void close()
 130     throws IOException {
 131         _sock.close();
 132     }
 133
 134     /**
 135      * Gets the socket to the PEP
 136      * @return   Socket connected to the PEP
 137      */
 138     public Socket getSocket() {
 139         return _sock;
 140     }
 141
 142     /**
 143      * Main loop
 144      */
 145     public void run () {
 146         Date _lastSendKa = new Date();
 147         _lastRecKa = new Date();
 148         try {
 149             while (!_sock.isClosed()) {
 150                 if (_sock.getInputStream().available() != 0) {
 151                     processMessage(_sock);
 152                     _lastRecKa = new Date();
 153                 }
 154
 155                 // Keep Alive
 156                 if (_kaTimer > 0) {
 157                     // Timeout at PDP
 158                     int _startTime = (int) (_lastRecKa.getTime());
 159                     int cTime = (int) (new Date().getTime());
 160
 161                     if ((cTime - _startTime) > _kaTimer*1000) {
 162                         _sock.close();
 163                         // Notify all Request State Managers
 164                         notifyNoKAAllReqStateMan();
 165                     }
 166
 167                     // Send to PEP
 168                     _startTime = (int) (_lastSendKa.getTime());
 169                     cTime = (int) (new Date().getTime());
 170
 171                     if ((cTime - _startTime) > ((_kaTimer*3/4)*1000)) {
 172                         // TODO - determine what is the client type to be used here?
 173                         final COPSKAMsg msg = new COPSKAMsg(ClientType.NA, null);
 174                         COPSTransceiver.sendMsg(msg, _sock);
 175                         _lastSendKa = new Date();
 176                     }
 177                 }
 178
 179                 try {
 180                     Thread.sleep(500);
 181                 } catch (Exception e) {
 182                     logger.error("Unexpected exception while sleeping", e);
 183                 }
 184
 185             }
 186         } catch (Exception e) {
 187             logger.error("Error reading messages from socket", e);
 188         }
 189
 190         // connection closed by server
 191         // COPSDebug.out(getClass().getName(),"Connection closed by client");
 192         try {
 193             _sock.close();
 194         } catch (IOException e) {
 195             logger.error("Error closing socket", e);
 196         }
 197
 198         // Notify all Request State Managers
 199         try {
 200             notifyCloseAllReqStateMan();
 201         } catch (COPSPdpException e) {
 202             logger.error("Error closing state managers", e);
 203         }
 204     }
 205
 206     /**
 207      * Gets a COPS message from the socket and processes it
 208      * @param    conn Socket connected to the PEP
 209      */
 210     private void processMessage(final Socket conn) throws COPSPdpException, COPSException, IOException {
 211         final COPSMsg msg = COPSTransceiver.receiveMsg(conn);
 212
 213         if (msg.getHeader().getOpCode().equals(OPCode.CC)) {
 214             handleClientCloseMsg(conn, msg);
 215         } else if (msg.getHeader().getOpCode().equals(OPCode.KA)) {
 216             handleKeepAliveMsg(conn, msg);
 217         } else if (msg.getHeader().getOpCode().equals(OPCode.REQ)) {
 218             handleRequestMsg(conn, msg);
 219         } else if (msg.getHeader().getOpCode().equals(OPCode.RPT)) {
 220             handleReportMsg(conn, msg);
 221         } else if (msg.getHeader().getOpCode().equals(OPCode.DRQ)) {
 222             handleDeleteRequestMsg(conn, msg);
 223         } else if (msg.getHeader().getOpCode().equals(OPCode.SSQ)) {
 224             handleSyncComplete(conn, msg);
 225         } else {
 226             throw new COPSPdpException("Message not expected (" + msg.getHeader().getOpCode() + ").");
 227         }
 228     }
 229
 230     /**
 231      * Handle Client Close Message, close the passed connection
 232      *
 233      * @param    conn                a  Socket
 234      * @param    msg                 a  COPSMsg
 235      *
 236      *
 237      * <Client-Close> ::= <Common Header>
 238      *  <Error>
 239      *  [<Integrity>]
 240      *
 241      * Not support [<Integrity>]
 242      *
 243      */
 244     private void handleClientCloseMsg(Socket conn, COPSMsg msg) {
 245         COPSClientCloseMsg cMsg = (COPSClientCloseMsg) msg;
 246         _error = cMsg.getError();
 247         try {
 248             // Support
 249             if (cMsg.getIntegrity() != null) {
 250                 logger.error("Unsupported objects (Integrity) to connection " + conn.getInetAddress());
 251             }
 252
 253             conn.close();
 254         } catch (Exception unae) {
 255             logger.error("Unexpected exception closing connection", unae);
 256         }
 257     }
 258
 259     /**
 260      * Gets the occurred COPS Error
 261      * @return   <tt>COPSError</tt> object
 262      */
 263     protected COPSError getError()  {
 264         return _error;
 265     }
 266
 267     /**
 268      * Handle Keep Alive Message
 269      *
 270      * <Keep-Alive> ::= <Common Header>
 271      *                  [<Integrity>]
 272      *
 273      * Not support [<Integrity>]
 274      *
 275      * @param    conn                a  Socket
 276      * @param    msg                 a  COPSMsg
 277      *
 278      */
 279     private void handleKeepAliveMsg(Socket conn, COPSMsg msg) {
 280         COPSKAMsg cMsg = (COPSKAMsg) msg;
 281
 282         COPSKAMsg kaMsg = (COPSKAMsg) msg;
 283         try {
 284             // Support
 285             if (cMsg.getIntegrity() != null) {
 286                 logger.error("Unsupported objects (Integrity) to connection " + conn.getInetAddress());
 287             }
 288
 289             kaMsg.writeData(conn);
 290         } catch (Exception unae) {
 291             logger.error("Unexpected exception while writing keep-alive message", unae);
 292         }
 293     }
 294
 295     /**
 296      * Handle Delete Request Message
 297      *
 298      * <Delete Request> ::= <Common Header>
 299      *                      <Client Handle>
 300      *                      <Reason>
 301      *                      [<Integrity>]
 302      *
 303      * Not support [<Integrity>]
 304      *
 305      * @param    conn                a  Socket
 306      * @param    msg                 a  COPSMsg
 307      *
 308      */
 309     private void handleDeleteRequestMsg(Socket conn, COPSMsg msg) throws COPSPdpException {
 310         COPSDeleteMsg cMsg = (COPSDeleteMsg) msg;
 311         // COPSDebug.out(getClass().getName(),"Removing ClientHandle for " +
 312         //  conn.getInetAddress() + ":" + conn.getPort() + ":[Reason " + cMsg.getReason().getDescription() + "]");
 313
 314         // Support
 315         if (cMsg.getIntegrity() != null) {
 316             logger.error("Unsupported objects (Integrity) to connection " + conn.getInetAddress());
 317         }
 318
 319         // Delete clientHandler
 320         if (_managerMap.remove(cMsg.getClientHandle().getId().str()) == null) {
 321             // TODO - Determine what to do here???
 322             // COPSDebug.out(getClass().getName(),"Missing for ClientHandle " +
 323             //  cMsg.getClientHandle().getId().getData());
 324         }
 325
 326         final PCMMPdpReqStateMan man = _managerMap.get(cMsg.getClientHandle().getId().str());
 327         if (man == null) {
 328             logger.warn("State manager not found");
 329         } else {
 330             man.processDeleteRequestState(cMsg);
 331         }
 332
 333     }
 334
 335     /**
 336      * Handle Request Message
 337      *
 338      * <Request> ::= <Common Header>
 339      *  <Client Handle>
 340      *  <Context>
 341      *  *(<Named ClientSI>)
 342      *  [<Integrity>]
 343      * <Named ClientSI> ::= <*(<PRID> <EPD>)>
 344      *
 345      * Not support [<Integrity>]
 346      *
 347      * @param    conn                a  Socket
 348      * @param    msg                 a  COPSMsg
 349      *
 350      */
 351     private void handleRequestMsg(Socket conn, COPSMsg msg) throws COPSPdpException {
 352
 353         final COPSReqMsg reqMsg = (COPSReqMsg) msg;
 354 //        COPSContext cntxt = reqMsg.getContext();
 355         final COPSHeader header = reqMsg.getHeader();
 356         //short reqType = cntxt.getRequestType();
 357         final ClientType cType = header.getClientType();
 358
 359         // Support
 360         if (reqMsg.getIntegrity() != null) {
 361             logger.error("Unsupported objects (Integrity) to connection " + conn.getInetAddress());
 362         }
 363
 364         PCMMPdpReqStateMan man;
 365         man = _managerMap.get(reqMsg.getClientHandle().getId().str());
 366         if (man == null) {
 367
 368             man = new PCMMPdpReqStateMan(cType, reqMsg.getClientHandle().getId().str());
 369             _managerMap.put(reqMsg.getClientHandle().getId().str(), man);
 370             man.setDataProcess(_process);
 371             man.initRequestState(_sock);
 372
 373             // COPSDebug.out(getClass().getName(),"createHandler called, clientType=" +
 374             //    header.getClientType() + " msgType=" +
 375             //    cntxt.getMessageType() + ", connId=" + conn.toString());
 376         }
 377
 378         man.processRequest(reqMsg);
 379     }
 380
 381     /**
 382      * Handle Report Message
 383      *
 384      * <Report State> ::= <Common Header>
 385      *  <Client Handle>
 386      *  <Report Type>
 387      *  *(<Named ClientSI>)
 388      *  [<Integrity>]
 389      *
 390      * Not support [<Integrity>]
 391      *
 392      * @param    conn                a  Socket
 393      * @param    msg                 a  COPSMsg
 394      *
 395      */
 396     private void handleReportMsg(Socket conn, COPSMsg msg)
 397     throws COPSPdpException {
 398         COPSReportMsg repMsg = (COPSReportMsg) msg;
 399         // COPSHandle handle = repMsg.getClientHandle();
 400         // COPSHeader header = repMsg.getHeader();
 401
 402         // Support
 403         if (repMsg.getIntegrity() != null) {
 404             logger.error("Unsupported objects (Integrity) to connection " + conn.getInetAddress());
 405         }
 406
 407         final PCMMPdpReqStateMan man = _managerMap.get(repMsg.getClientHandle().getId().str());
 408         if (man == null) {
 409             logger.warn("State manager not found");
 410         } else {
 411             man.processReport(repMsg);
 412         }
 413     }
 414
 415     /**
 416      * Method handleSyncComplete
 417      *
 418      * @param    conn                a  Socket
 419      * @param    msg                 a  COPSMsg
 420      *
 421      */
 422     private void handleSyncComplete(Socket conn, COPSMsg msg)
 423     throws COPSPdpException {
 424         COPSSyncStateMsg cMsg = (COPSSyncStateMsg) msg;
 425         // COPSHandle handle = cMsg.getClientHandle();
 426         // COPSHeader header = cMsg.getHeader();
 427
 428         // Support
 429         if (cMsg.getIntegrity() != null) {
 430             logger.error("Unsupported objects (Integrity) to connection " + conn.getInetAddress());
 431         }
 432
 433         final PCMMPdpReqStateMan man = _managerMap.get(cMsg.getClientHandle().getId().str());
 434         if (man == null) {
 435             logger.warn("State manager not found");
 436         } else {
 437             man.processSyncComplete(cMsg);
 438         }
 439     }
 440
 441     /**
 442      * Requests a COPS sync from the PEP
 443      * @throws COPSException
 444      * @throws COPSPdpException
 445      */
 446     protected void syncAllRequestState() throws COPSException, COPSPdpException {
 447         for (final PCMMPdpReqStateMan man : _managerMap.values()) {
 448             man.syncRequestState();
 449         }
 450     }
 451
 452     private void notifyCloseAllReqStateMan() throws COPSPdpException {
 453         for (final PCMMPdpReqStateMan man : _managerMap.values()) {
 454             man.processClosedConnection(_error);
 455         }
 456     }
 457
 458     private void notifyNoKAAllReqStateMan() throws COPSPdpException {
 459         for (final PCMMPdpReqStateMan man : _managerMap.values()) {
 460             man.processNoKAConnection();
 461         }
 462     }
 463
 464 }