gt:                     eventRaw.address = buffer.getInt();
context: /*
   * AEUnicastDialog.java
   *
   * Created on April 25, 2008, 8:40 AM
   */
  package net.sf.jaer.eventio;
  
  import com.jogamp.opengl.util.GLDrawableUtil;
  import java.beans.PropertyChangeEvent;
  import java.beans.PropertyChangeListener;
  import java.io.IOException;
  import java.net.DatagramSocket;
  import java.net.InetSocketAddress;
  import java.net.SocketAddress;
  import java.net.SocketTimeoutException;
  import java.nio.BufferOverflowException;
  import java.nio.ByteBuffer;
  import java.nio.ByteOrder;
  import java.nio.MappedByteBuffer;
  import java.nio.channels.DatagramChannel;
  import java.util.concurrent.ArrayBlockingQueue;
  import java.util.concurrent.Semaphore;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  import java.util.prefs.Preferences;
  
  import net.sf.jaer.aemonitor.AENetworkRawPacket;
  import net.sf.jaer.aemonitor.AEPacket;
  import net.sf.jaer.aemonitor.AEPacketRaw;
  import net.sf.jaer.aemonitor.EventRaw;
  import net.sf.jaer.aemonitor.EventRaw.EventType;
  import net.sf.jaer.chip.AEChip;
  import net.sf.jaer.chip.EventExtractor2D;
  import net.sf.jaer.event.EventPacket;
  import static net.sf.jaer.eventio.AEFileInputStream.MAX_BUFFER_SIZE_EVENTS;
  import net.sf.jaer.eventio.Jaer3BufferParser;
  import net.sf.jaer.eventio.Jaer3BufferParser.PacketDescriptor;
  import net.sf.jaer.eventio.Jaer3BufferParser.Jaer3EventExtractor;
  
  /**
   * Receives input via datagram (connectionless, UDP) packets from a server.
   * <p>
   * The socket binds to the port which comes initially from the Preferences for
   * AEUnicastInput. The port can be later changed.
   * <p>
   * Each packet consists of (by default)
   *
   * 1. a packet sequence integer (32 bits) which can be used to count missed
   * packets
   *
   * 2. AEs. Each AE is a pair int32 address, int32 timestamp. Timestamps are
   * assumed to have 1us tick.
   * <p>
   * Options allow different choices for use of sequence number, size of
   * address/timestamp, order of address/timestamp, and swapping byte order to
   * account for big/little endian peers.
   *
   * <p>
   * The datagram socket is not connected to the receiver, i.e., connect() is not
   * called on the socket.
   *
   * @see #setAddressFirstEnabled
   * @see #setSequenceNumberEnabled
   *
   *
   */
  public class AEUnicastInput implements AEUnicastSettings, PropertyChangeListener {
  
      private int NBUFFERS = 100; // should match somehow the expected number of datagrams that come in a burst before the readPacket() method is called.
  
      // TODO If the remote host sends 16 bit timestamps, then a local unwrapping is done to extend the time range
      private static Preferences prefs = Preferences.userNodeForPackage(AEUnicastInput.class);
      private DatagramSocket datagramSocket = null;
      private boolean printedHost = false;
      private int port = prefs.getInt("AEUnicastInput.port", AENetworkInterfaceConstants.DATAGRAM_PORT);
      private boolean sequenceNumberEnabled = prefs.getBoolean("AEUnicastInput.sequenceNumberEnabled", true);
      private boolean cAERStreamEnabled = prefs.getBoolean("AEUnicastInput.cAERDisplayEnabled", true);
      private boolean addressFirstEnabled = prefs.getBoolean("AEUnicastInput.addressFirstEnabled", true);
      private ArrayBlockingQueue<ByteBuffer> filledBufferQueue = new ArrayBlockingQueue(NBUFFERS), availableBufferQueue = new ArrayBlockingQueue(NBUFFERS);
      private AENetworkRawPacket packet = new AENetworkRawPacket();
      private static final Logger log = Logger.getLogger("AESocketStream");
      private int bufferSize = prefs.getInt("AEUnicastInput.bufferSize", AENetworkInterfaceConstants.DATAGRAM_BUFFER_SIZE_BYTES);
      private boolean swapBytesEnabled = prefs.getBoolean("AEUnicastInput.swapBytesEnabled", false);
      private float timestampMultiplier = prefs.getFloat("AEUnicastInput.timestampMultiplier", DEFAULT_TIMESTAMP_MULTIPLIER);
      private boolean use4ByteAddrTs = prefs.getBoolean("AEUnicastInput.use4ByteAddrTs", DEFAULT_USE_4_BYTE_ADDR_AND_TIMESTAMP);
      private boolean localTimestampsEnabled = prefs.getBoolean("AEUnicastOutput.localTimestampsEnabled", false);
      private boolean spinnakerProtocolEnabled = prefs.getBoolean("AEUnicastInput.spinnakerProtocolEnabled", false);
      private boolean secDvsProtocolEnabled = prefs.getBoolean("AEUnicastInput.secDvsProtocolEnabled", false);
      boolean stopme = false;
      private DatagramChannel channel;
      private int datagramCounter = 0;
      private int datagramSequenceNumber = 0;
      private EventRaw eventRaw = new EventRaw();
      private int timeZero = 0; // used to store initial timestamp for 4 byte timestamp reads to subtract this value
      private boolean readTimeZeroAlready = true;
      private boolean timestampsEnabled = prefs.getBoolean("AEUnicastInput.timestampsEnabled", DEFAULT_TIMESTAMPS_ENABLED);
      private Semaphore pauseSemaphore = new Semaphore(1);
      private volatile boolean paused = false;
      private Reader readingThread = null;
      private AEChip chip = null; // needed to support cAER jaer3.0 decoding to jAER format
      private EventExtractor2D restoreEventExtractor = null; // The restore extractor
      private ByteBuffer wholePktBuffer = null;
      private int jaer3PktSize = 0, jaer3PktNum = 0;
      private long jaer3EventsNum = 0;
      private Jaer3BufferParser j3Parser;
      private int secGen2TimestampMSB = 0;
      private int secGen2TimestampLSB = 0;
  
      /**
       * Constructs an instance of AEUnicastInput and binds it to the default
       * port. The port preference value may have been modified from the
       * Preferences default by a previous setPort() call which stored the
       * preference value.
       * <p>
       * This Thread subclass must be started in order to receive event packets.
       *
       * @see AENetworkInterfaceConstants
       *
       */
      public AEUnicastInput(AEChip chip) { // TODO basic problem here is that if port is unavailable, then we cannot construct and set port
          this.chip = chip;
          restoreEventExtractor = chip.getEventExtractor();
      }
  
      /**
       * Constructs a new AEUnicastInput using the given port number.
       *
       * @param port the UDP port number.
       * @see AEUnicastInput#AEUnicastInput()
       */
      public AEUnicastInput(int port, AEChip chip) {
          this(chip);
          setPort(port);
      }
  
      private void allocateBufffers() {
          availableBufferQueue.clear();
          for (int i = 0; i < NBUFFERS; i++) {
              ByteBuffer buffer = ByteBuffer.allocateDirect(bufferSize);
              buffer.order(swapBytesEnabled || spinnakerProtocolEnabled || secDvsProtocolEnabled ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN); //spinnaker always uses little endian
              availableBufferQueue.add(buffer);
          }
          filledBufferQueue.clear();
      }
  
      private void freeBuffers() {
          availableBufferQueue.clear();
          filledBufferQueue.clear(); // allow GC to collect these references
      }
  
      private int eventSize() {
          if (use4ByteAddrTs) {
              if (timestampsEnabled) {
                  return AENetworkInterfaceConstants.EVENT_SIZE_BYTES;
              } else {
                  return 4;
              }
          } else if (timestampsEnabled) {
              return 4;
          } else {
              return 2;
          }
      }
  
      /**
       * Returns the latest buffer of events. If a timeout occurs occurs a null
       * packet is returned.
       *
       * @return the events collected since the last call to readPacket(), or null
       * on a timeout or interrupt.
       */
      int nTmp = 0;
      int nEventCapacity = 0;
  
      public AENetworkRawPacket readPacket() {
          packet.clear();
          readingThread.maxSizeExceeded = false;
          try {
              returnearly:
              while (filledBufferQueue.peek() != null) {
                  ByteBuffer buffer = filledBufferQueue.take();
                  // buffer.clear();
                  extractEvents(buffer, packet);
                  buffer.clear();
                  availableBufferQueue.put(buffer);
                  if (packet.getNumEvents() >= 10000) {
                      break returnearly; // Set a threshold to avoid the big dealy caused by accumulating too many events in the packet.
                  }
              }
              return packet;
          } catch (InterruptedException e) {
              log.info("Interrupted exchange of buffers in AEUnicastInput: " + e.toString());
              return null;
          }
      }
  
      private void checkSequenceNumber(ByteBuffer buffer) {
          if (sequenceNumberEnabled) {
              datagramSequenceNumber = buffer.getInt(); // swab(buffer.getInt());
  //                log.info("recieved packet with sequence number "+packetSequenceNumber);
              if (datagramSequenceNumber != datagramCounter) {
                  log.warning(String.format("Dropped %d packets. (Incoming packet sequence number (%d) doesn't match expected packetCounter (%d), resetting packetCounter)", datagramSequenceNumber - datagramCounter, datagramSequenceNumber, datagramCounter));
                  datagramCounter = datagramSequenceNumber;
              }
              datagramCounter++;
          }
      }
  
      /**
       * Receives a buffer from the UDP socket. Data is stored in internal buffer.
       *
       * @param packet used to set number of events to 0 if there is an error and
       * to store the source host(s) information.
       * @return client if successful, null if there is an IOException.
       */
      private SocketAddress receiveDatagramAndPutToExchanger(AENetworkRawPacket packet) {
          SocketAddress client = null;
          try {
              ByteBuffer buffer = availableBufferQueue.take(); // buffer must be cleared by readPacket
  //            ByteBuffer buffer = ByteBuffer.allocateDirect(bufferSize);
  //            buffer.order(swapBytesEnabled ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN);
  
              client = channel.receive(buffer); // fill buffer with data from datagram, blocks here until packet received
              if (!printedHost) {
                  printedHost = true;
                  log.info("received first packet from " + client + " of length " + buffer.position() + " bytes"); // , connecting channel
                  // do not connect so that multiple clients can send us data on the same port
                  //                channel.connect(client); // TODO connect the channel to avoid security check overhead. TODO check if handles reconnects correctly.
              }
              if (client instanceof InetSocketAddress) {
                  packet.addClientAddress((InetSocketAddress) client, packet.getNumEvents());
              } else if (client == null) {
                  paused = true;
                  log.warning("Device not connected or wrong configured. Datagrams have to be sent to port: " + port + " .Input stream paused.");
              } else {
                  log.warning("unknown type of client address - should be InetSocketAddress: " + client);
              }
              buffer.flip();
              if (!spinnakerProtocolEnabled && !secDvsProtocolEnabled) {
                  checkSequenceNumber(buffer);
              }
  //            if(exchanger.size()>10){
  //                log.info("filled queue of datagrams has "+exchanger.size()+" buffers");
  //            }
              filledBufferQueue.put(buffer); // blocks here until readPacket clears the packet
          } catch (InterruptedException ie) {
              log.warning(ie.toString());
              return null;
          } catch (SocketTimeoutException to) {
              // just didn't fill the buffer in time, ignore
              log.warning(to.toString());
              return null;
          } catch (IOException e) {
              log.warning(e.toString());
              packet.clear();
              return null;
          } catch (IllegalArgumentException eArg) {
              log.warning(eArg.toString());
              return null;
          }
          return client;
      }
  
      private int maybeSwapByteOrder(int value) {
          if (!swapBytesEnabled) {
              return value;
          }
          int b1 = (value >> 0) & 0xff;
          int b2 = (value >> 8) & 0xff;
          int b3 = (value >> 16) & 0xff;
          int b4 = (value >> 24) & 0xff;
  
          return b1 << 24 | b2 << 16 | b3 << 8 | b4 << 0;
      }
  
      private short maybeSwapByteOrder(short value) {
          if (!swapBytesEnabled) {
              return value;
          }
          int b1 = (value >> 0) & 0xff;
          int b2 = (value >> 8) & 0xff;
  
          return (short) ((0xffff & (b1 << 8) | (0xffff & (b2 << 0))));
      }
  
      private static ByteBuffer clone(ByteBuffer original) {
          ByteBuffer clone = ByteBuffer.allocate(original.capacity());
          if (clone.capacity() < 90000) { // The max length of a whole frame packet is 86464, so the capacity must be bigger than 86464, here we use 90000 for convience. 
              clone = ByteBuffer.allocate(90000);
          }
          original.rewind(); //copy from the beginning
          clone.put(original);
          original.rewind();
          // clone.flip(); // We still need to put data in the buffer, we will flip it in the end buffer, so we don't need flip at the head buffer.
          return clone;
      }
  
      /**
       * Extracts data from internal buffer and adds to packet, according to all
       * the option flags.
       *
       * @param packet to add events to.
       */
      private void extractEvents(ByteBuffer buffer, AENetworkRawPacket packet) {
  
          if (isSpinnakerProtocolEnabled()) {
              // Read header
              byte byte0 = buffer.get();
              if ((byte0 & 0b10000000) != 0) {
                  throw new UnsupportedOperationException("Command packets not supported yet.");
              }
              if (use4ByteAddrTs != ((byte0 & 0b01000000) != 0)) {
                  throw new IllegalStateException("use4ByteAddrTs conflict btw header and user setting.");
              }
              if ((!timestampsEnabled || localTimestampsEnabled) != ((byte0 & 0b00100000) == 0)) {
                  throw new IllegalStateException("timestamp enabling conflict btw header and user setting.");
              }
              //we ignore payloads, but need their sizes to skip them
              int payload_size; //payload size in bytes
              {
                  boolean bit1, bit0;
                  bit1 = ((byte0 & 0b00010000) != 0);
                  bit0 = ((byte0 & 0b00001000) != 0);
                  if (!bit0 && !bit1) {
                      payload_size = 0; //no payload
                  } else if (bit0 && !bit1) {
                      payload_size = 2; //16bit
                  } else if (!bit0 && bit1) {
                      payload_size = 4; //32bit
                  } else {
                      payload_size = 16; //128bit
                  }
              }
              //if(payload_size != 0) throw new UnsupportedOperationException("Payloads not supported.");
              //key prefix
              if ((byte0 & 0b00000100) != 0) {
                  throw new UnsupportedOperationException("Key prefixes are not supported.");
              }
              //timestamp prefix
              if ((byte0 & 0b00000010) != 0) {
                  throw new UnsupportedOperationException("Timestamp prefixes are not supported.");
              }
              //payload prefix
              if ((byte0 & 0b00000001) != 0) {
                  throw new UnsupportedOperationException("Payload prefixes are not supported.");
              }
  
              //read number of events
              int nEventsInPacket = buffer.get() & 0xff;
              int eventSize = (2 * (use4ByteAddrTs ? 4 : 2)) + payload_size;
              int computednEventsInPacket = (buffer.limit() - 4) / eventSize;
              if (computednEventsInPacket != nEventsInPacket) {
                  nEventsInPacket = Math.min(computednEventsInPacket, nEventsInPacket);
                  log.warning("Mismatch between number of events claimed by header and the computed one from packet size. Using smallest one.");
              }
  
              //packet counter
              int packetNumber = buffer.get() & 0xff;
              if (sequenceNumberEnabled) {
                  datagramSequenceNumber = packetNumber; // swab(buffer.getInt());
                  //                log.info("recieved packet with sequence number "+packetSequenceNumber);
                  if (datagramSequenceNumber != datagramCounter) {
                      log.warning(String.format("Dropped %d packets. (Incoming packet sequence number (%d) doesn't match expected packetCounter (%d), resetting packetCounter)", datagramSequenceNumber - datagramCounter, datagramSequenceNumber, datagramCounter));
                      datagramCounter = datagramSequenceNumber;
                  }
                  if (datagramCounter < 255) {
                      datagramCounter++;
                  } else {
                      datagramCounter = 0;
                  }
              }
  
              //Reserved byte, just ignore it
              buffer.get();
  
              //read events
              int ts = !timestampsEnabled || localTimestampsEnabled ? (int) (((System.nanoTime() / 1000) << 32) >> 32) : 0; // if no timestamps coming, add system clock for all.
              final int startingIndex = packet.getNumEvents();
              final int newPacketLength = startingIndex + nEventsInPacket;
              packet.ensureCapacity(newPacketLength);
              final int[] addresses = packet.getAddresses();
              final int[] timestamps = packet.getTimestamps();
  
              for (int i = 0; i < nEventsInPacket; i++) {
                  if (use4ByteAddrTs) {
  