protocol_plugins.packetcable/src/main/java/org/pcmm/nio/PCMMChannel.java

   1 /**
   2  *
   3  */
   4 package org.pcmm.nio;
   5
   6 import java.io.IOException;
   7 import java.io.InputStream;
   8 import java.net.Socket;
   9 import java.nio.ByteBuffer;
  10 import java.util.Date;
  11
  12 import org.pcmm.PCMMProperties;
  13 import org.slf4j.Logger;
  14 import org.slf4j.LoggerFactory;
  15 import org.umu.cops.stack.COPSException;
  16 import org.umu.cops.stack.COPSHeader;
  17 import org.umu.cops.stack.COPSMsg;
  18 import org.umu.cops.stack.COPSMsgParser;
  19
  20 /**
  21  * this class provides a set of utilities to efficiently read/write data from a
  22  * stream, it could parameterized with a reading timeout or -1 for blocking
  23  * until get a message
  24  *
  25  */
  26 public class PCMMChannel {
  27
  28         private Logger logger = LoggerFactory.getLogger(getClass().getName());
  29         private ByteBuffer dataBuffer;
  30         private Socket socket;
  31         private int timeout;
  32         public static final int DEFAULT_BYTE_BUFFER_SIZE = 2048;
  33         public static final int DEFAULT_READ_TIMEOUT = -1;
  34
  35         public PCMMChannel(Socket socket) {
  36                 this(socket, PCMMProperties.get(PCMMProperties.DEFAULT_TIEMOUT,
  37                                 Integer.class, DEFAULT_READ_TIMEOUT));
  38         }
  39
  40         public PCMMChannel(Socket socket, int timeout) {
  41                 this.socket = socket;
  42                 dataBuffer = ByteBuffer.allocateDirect(DEFAULT_BYTE_BUFFER_SIZE);
  43                 logger.info("Allocated byte buffer with size = "
  44                                 + DEFAULT_BYTE_BUFFER_SIZE);
  45                 this.timeout = timeout;
  46                 logger.info("Set read/write timeout to : " + timeout);
  47
  48         }
  49
  50         public int readData(byte[] dataRead, int nchar) throws IOException {
  51                 InputStream input;
  52                 input = getSocket().getInputStream();
  53                 int nread = 0;
  54                 int startTime = (int) (new Date().getTime());
  55                 do {
  56                         if (timeout == -1 || input.available() != 0) {
  57                                 nread += input.read(dataRead, nread, nchar - nread);
  58                                 startTime = (int) (new Date().getTime());
  59                         } else {
  60                                 int nowTime = (int) (new Date().getTime());
  61                                 if ((nowTime - startTime) > timeout)
  62                                         break;
  63                         }
  64                 } while (nread != nchar);
  65                 return nread;
  66         }
  67
  68         /**
  69          * Method sendMsg
  70          *
  71          * @param msg
  72          *            a COPSMsg
  73          *
  74          * @throws IOException
  75          * @throws COPSException
  76          *
  77          */
  78         public void sendMsg(COPSMsg msg) throws IOException, COPSException {
  79                 logger.debug("sendMsg({})==>{}", getSocket(), msg);
  80                 msg.checkSanity();
  81                 msg.writeData(getSocket());
  82         }
  83
  84         /**
  85          * Method receiveMessage
  86          *
  87          * @return a COPSMsg
  88          *
  89          * @throws IOException
  90          * @throws COPSException
  91          *
  92          */
  93         public COPSMsg receiveMessage() throws IOException, COPSException {
  94                 int nread = 0;
  95                 byte[] hBuf = new byte[8];
  96
  97                 logger.debug("receiveMessage({})", getSocket());
  98
  99                 nread = readData(hBuf, 8);
 100
 101                 if (nread == 0) {
 102                         throw new COPSException("Error reading connection");
 103                 }
 104
 105                 if (nread != 8) {
 106                         throw new COPSException("Bad COPS message");
 107                 }
 108
 109                 COPSHeader hdr = new COPSHeader(hBuf);
 110                 int dataLen = hdr.getMsgLength() - hdr.getHdrLength();
 111                 logger.debug("COPS Msg length :[" + dataLen + "]\n");
 112                 byte[] buf = new byte[dataLen + 1];
 113                 nread = 0;
 114
 115                 nread = readData(buf, dataLen);
 116                 buf[dataLen] = (byte) '\0';
 117                 logger.debug("Data read length:[" + nread + "]\n");
 118
 119                 if (nread != dataLen) {
 120                         throw new COPSException("Bad COPS message");
 121                 }
 122                 COPSMsgParser prser = new COPSMsgParser();
 123                 COPSMsg msg = prser.parse(hdr, buf);
 124                 return msg;
 125         }
 126
 127         /**
 128          * @return the dataBuffer
 129          */
 130         public ByteBuffer getDataBuffer() {
 131                 return dataBuffer;
 132         }
 133
 134         /**
 135          * @param dataBuffer
 136          *            the dataBuffer to set
 137          */
 138         public void setDataBuffer(ByteBuffer dataBuffer) {
 139                 this.dataBuffer = dataBuffer;
 140         }
 141
 142         /**
 143          * @return the socket
 144          */
 145         public Socket getSocket() {
 146                 return socket;
 147         }
 148
 149         /**
 150          * @param socket
 151          *            the socket to set
 152          */
 153         public void setSocket(Socket socket) {
 154                 this.socket = socket;
 155         }
 156
 157         /**
 158          * @return the timeout
 159          */
 160         public int getTimeout() {
 161                 return timeout;
 162         }
 163
 164         /**
 165          * @param timeout
 166          *            the timeout to set
 167          */
 168         public void setTimeout(int timeout) {
 169                 this.timeout = timeout;
 170         }
 171
 172 }