third-party/openflowj/src/main/java/org/openflow/io/OFMessageAsyncStream.java

   1 /**
   2  *
   3  */
   4 package org.openflow.io;
   5
   6 import java.io.IOException;
   7 import java.nio.ByteBuffer;
   8 import java.nio.channels.SocketChannel;
   9 import java.util.List;
  10
  11 import org.openflow.protocol.OFMessage;
  12 import org.openflow.protocol.factory.OFMessageFactory;
  13
  14 /**
  15  * Asynchronous OpenFlow message marshalling and unmarshalling stream wrapped
  16  * around an NIO SocketChannel
  17  *
  18  * @author Rob Sherwood (rob.sherwood@stanford.edu)
  19  * @author David Erickson (daviderickson@cs.stanford.edu)
  20  *
  21  */
  22 public class OFMessageAsyncStream implements OFMessageInStream,
  23         OFMessageOutStream {
  24     static public int defaultBufferSize = 1048576; // 1MB
  25
  26     protected ByteBuffer inBuf, outBuf;
  27     protected OFMessageFactory messageFactory;
  28     protected SocketChannel sock;
  29     protected int partialReadCount = 0;
  30
  31     public OFMessageAsyncStream(SocketChannel sock,
  32             OFMessageFactory messageFactory) throws IOException {
  33         inBuf = ByteBuffer
  34                 .allocateDirect(OFMessageAsyncStream.defaultBufferSize);
  35         outBuf = ByteBuffer
  36                 .allocateDirect(OFMessageAsyncStream.defaultBufferSize);
  37         this.sock = sock;
  38         this.messageFactory = messageFactory;
  39         this.sock.configureBlocking(false);
  40     }
  41
  42     @Override
  43     public List<OFMessage> read() throws IOException {
  44         return this.read(0);
  45     }
  46
  47     @Override
  48     public List<OFMessage> read(int limit) throws IOException {
  49         List<OFMessage> l;
  50         int read = sock.read(inBuf);
  51         if (read == -1)
  52             return null;
  53         inBuf.flip();
  54         l = messageFactory.parseMessages(inBuf, limit);
  55         if (inBuf.hasRemaining())
  56             inBuf.compact();
  57         else
  58             inBuf.clear();
  59         return l;
  60     }
  61
  62     protected void appendMessageToOutBuf(OFMessage m) throws IOException {
  63         int msglen = m.getLengthU();
  64         if (outBuf.remaining() < msglen) {
  65             throw new IOException(
  66                     "Message length exceeds buffer capacity: " + msglen);
  67         }
  68         m.writeTo(outBuf);
  69     }
  70
  71     /**
  72      * Buffers a single outgoing openflow message
  73      */
  74     @Override
  75     public void write(OFMessage m) throws IOException {
  76         appendMessageToOutBuf(m);
  77     }
  78
  79     /**
  80      * Buffers a list of OpenFlow messages
  81      */
  82     @Override
  83     public void write(List<OFMessage> l) throws IOException {
  84         for (OFMessage m : l) {
  85             appendMessageToOutBuf(m);
  86         }
  87     }
  88
  89     /**
  90      * Flush buffered outgoing data. Keep flushing until needsFlush() returns
  91      * false. Each flush() corresponds to a SocketChannel.write(), so this is
  92      * designed for one flush() per select() event
  93      */
  94     public void flush() throws IOException {
  95         outBuf.flip(); // swap pointers; lim = pos; pos = 0;
  96         sock.write(outBuf); // write data starting at pos up to lim
  97         outBuf.compact();
  98     }
  99
 100     /**
 101      * Is there outgoing buffered data that needs to be flush()'d?
 102      */
 103     public boolean needsFlush() {
 104         return outBuf.position() > 0;
 105     }
 106
 107     /**
 108      * @return the messageFactory
 109      */
 110     public OFMessageFactory getMessageFactory() {
 111         return messageFactory;
 112     }
 113
 114     /**
 115      * @param messageFactory
 116      *            the messageFactory to set
 117      */
 118     public void setMessageFactory(OFMessageFactory messageFactory) {
 119         this.messageFactory = messageFactory;
 120     }
 121 }