if (isSender() && (startTime + myUpdateInterval / 2.0) >= myNextUpdate) {
if (mySocket == null)
// If for some reason the socket hasn't been initialized, then initialize it.
initialize();
if (myTerminations.isEmpty())
throw new SimulationException("SocketUDPNode is sender, but has no terminations to get data from.");
else {
// TODO: Test with spiking outputs?
float[] values = new float[myDimension];
Iterator<PassthroughTermination> it = myTerminations.values().iterator();
while (it.hasNext()) {
PassthroughTermination termination = it.next();
InstantaneousOutput io = termination.getValues();
if (io instanceof RealOutput) {
values = MU.sum(values, ((RealOutput) io).getValues());
} else if (io instanceof SpikeOutput) {
boolean[] spikes = ((SpikeOutput) io).getValues();
for (int i = 0; i < spikes.length; i++) {
if (spikes[i]) {
values[i] += 1f/(endTime - startTime);
}
}
} else if (io == null) {
throw new SimulationException("Null input to Termination " + termination.getName());
} else {
throw new SimulationException("Output type unknown: " + io.getClass().getName());
}
}
// Send values over the socket.
// Datagram format:
// - bytes 1-4: Timestamp (float)
// - bytes 4-(myDim+1)*4: values[i] (float)
ByteBuffer buffer = ByteBuffer.allocate((myDimension + 1) * 4);
buffer.order(myByteOrder);
buffer.putFloat((float)((startTime + endTime + myUpdateInterval) / 2.0));
for(int i = 0; i < myDimension; i++)
buffer.putFloat(values[i]);
byte[] bufArray = buffer.array();
DatagramPacket packet = new DatagramPacket(bufArray, bufArray.length, myDestAddress, myDestPort);
try {
mySocket.send(packet);
}
catch (IOException e) {
// TODO: Handle this better
throw new SimulationException(e);
}
}
}
if (isReceiver()) {
float[] values = new float[myOrigin.getDimensions()];
float[] tempValues = new float[myOrigin.getDimensions()+1];
// Check items in priority queue to see if there is anything that is good to go (i.e. timestamp is within
// startTime and endTime.
boolean foundItem = false;
boolean foundFutureItem = false;
int i = 0;
while (!mySocketBuffer.isEmpty()) {
tempValues = (float[]) mySocketBuffer.peek();
foundItem = (tempValues[0] >= startTime && tempValues[0] <= endTime) || myIgnoreTimestamp;
foundFutureItem = tempValues[0] > endTime;
if (foundItem)
mySocketBuffer.remove();
else
break;
}
if (foundItem) {
// If an item was found in the queue (i.e. message with future timestamp was received before), use this
// data instead of waiting on socket for new data.
for (i = 0; i < myOrigin.getDimensions(); i++)
values[i] = tempValues[i+1];
}
else if (foundFutureItem || startTime < myNextUpdate) {
// Buffer contained item in the future, so skip waiting for a new packet to arrive, and hurry
// the heck up.
values = ((RealOutputImpl) myOrigin.getValues()).getValues().clone();
}
else {
// If no items were found in queue, wait on socket for new data.
try {
byte[] bufArray = new byte[(myOrigin.getDimensions() + 1) * 4];
DatagramPacket packet = new DatagramPacket(bufArray, bufArray.length);
while (true) {
mySocket.receive(packet);
ByteBuffer buffer = ByteBuffer.wrap(bufArray);
buffer.order(myByteOrder);
for (i = 0; i < myOrigin.getDimensions() + 1; i++) {
tempValues[i] = buffer.getFloat();
}
// Check for timestamp for validity (i.e. within the start and end of this run call).
if ((tempValues[0] >= startTime && tempValues[0] <= endTime) || myIgnoreTimestamp || tempValues[0] < 0) {
// Valid timestamp encountered; or have been instructed to ignore timestamps.
// No further actions required, just break out of while loop.
System.arraycopy(tempValues, 1, values, 0, myOrigin.getDimensions());
break;
}
else if (tempValues[0] > endTime) {
// Future timestamp encountered, place into priority queue, use previous origin value as
// current value, and then out of while loop.
// Note: we break out of the while loop because receiving future timestamps means this
// system is (potentially) running slow.
mySocketBuffer.add(tempValues.clone());
values = ((RealOutputImpl) myOrigin.getValues()).getValues().clone();
break;
}
// Past timestamp encountered. Just ignore it, and wait for another packet.
}
}
catch (SocketTimeoutException e){
// If a timeout occurs, don't really do anything, just keep the origin at the previous value.
values = ((RealOutputImpl) myOrigin.getValues()).getValues().clone();
}
catch (Exception e){
// TODO: Handle this better
throw new SimulationException(e);
}
}
myOrigin.setValues(new RealOutputImpl(values, Units.UNK, endTime));
}
if (startTime >= myNextUpdate)