FastPublisher extends Kryo implements ClientListener, Stream { @Override @SuppressWarnings("SleepWhileInLoop") public void put(Object tuple) { if (tuple instanceof Tuple) { final Tuple t = (Tuple)tuple; byte[] array; switch (t.getType()) { case CHECKPOINT: array = WindowIdTuple.getSerializedTuple((int)t.getWindowId()); array[0] = MessageType.CHECKPOINT_VALUE; break; case BEGIN_WINDOW: array = BeginWindowTuple.getSerializedTuple((int)t.getWindowId()); break; case END_WINDOW: array = EndWindowTuple.getSerializedTuple((int)t.getWindowId()); break; case CUSTOM_CONTROL: array = null; break; case END_STREAM: array = EndStreamTuple.getSerializedTuple((int)t.getWindowId()); break; case RESET_WINDOW: com.datatorrent.stram.tuple.ResetWindowTuple rwt = (com.datatorrent.stram.tuple.ResetWindowTuple)t; array = ResetWindowTuple.getSerializedTuple(rwt.getBaseSeconds(), rwt.getIntervalMillis()); break; default: throw new UnsupportedOperationException("this data type is not handled in the stream"); } int size = array.length; if (writeBuffer.hasRemaining()) { writeBuffer.put((byte)size); if (writeBuffer.hasRemaining()) { writeBuffer.put((byte)(size >> 8)); } else { synchronized (readBuffers) { readBuffers[writeIndex].limit(BUFFER_CAPACITY); } advanceWriteBuffer(); writeBuffer.put((byte)(size >> 8)); } } else { synchronized (readBuffers) { readBuffers[writeIndex].limit(BUFFER_CAPACITY); } advanceWriteBuffer(); writeBuffer.put((byte)size); writeBuffer.put((byte)(size >> 8)); } int remaining = writeBuffer.remaining(); if (remaining < size) { int offset = 0; do { writeBuffer.put(array, offset, remaining); offset += remaining; size -= remaining; synchronized (readBuffers) { readBuffers[writeIndex].limit(BUFFER_CAPACITY); } advanceWriteBuffer(); remaining = writeBuffer.remaining(); if (size <= remaining) { writeBuffer.put(array, offset, size); break; } } while (true); } else { writeBuffer.put(array); } synchronized (readBuffers) { readBuffers[writeIndex].limit(writeBuffer.position()); } } else { count++; int hashcode = tuple.hashCode(); int wi = writeIndex; int position = writeBuffer.position(); int newPosition = position + 2 + 1 + 4 ; if (newPosition > BUFFER_CAPACITY) { writeBuffer.position(BUFFER_CAPACITY); advanceWriteBuffer(); writeBuffer.position(newPosition - BUFFER_CAPACITY); } else { writeBuffer.position(newPosition); } writeClassAndObject(output, tuple); int size; if (wi == writeIndex) { size = writeBuffer.position() - position - 2 ; assert (size <= Short.MAX_VALUE); writeBuffer.put(position++, (byte)size); writeBuffer.put(position++, (byte)(size >> 8)); writeBuffer.put(position++, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); writeBuffer.put(position++, (byte)hashcode); writeBuffer.put(position++, (byte)(hashcode >> 8)); writeBuffer.put(position++, (byte)(hashcode >> 16)); writeBuffer.put(position, (byte)(hashcode >> 24)); synchronized (readBuffers[wi]) { readBuffers[wi].limit(writeBuffer.position()); } } else { size = BUFFER_CAPACITY - position - 2 + writeBuffer.position(); int index = writeIndex; synchronized (readBuffers[index]) { readBuffers[index].position(0); readBuffers[index].limit(writeBuffer.position()); } do { if (index == 0) { index = lastIndex; } else { index--; } if (index == wi) { break; } synchronized (readBuffers[index]) { readBuffers[index].position(0); readBuffers[index].limit(BUFFER_CAPACITY); } size += BUFFER_CAPACITY; } while (true); assert (size <= Short.MAX_VALUE); index = wi; switch (position) { case BUFFER_CAPACITY: position = 0; if (wi == lastIndex) { wi = 0; } else { wi++; } writeBuffers[wi].put(position++, (byte)size); writeBuffers[wi].put(position++, (byte)(size >> 8)); writeBuffers[wi].put(position++, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); writeBuffers[wi].put(position++, (byte)hashcode); writeBuffers[wi].put(position++, (byte)(hashcode >> 8)); writeBuffers[wi].put(position++, (byte)(hashcode >> 16)); writeBuffers[wi].put(position, (byte)(hashcode >> 24)); break; case BUFFER_CAPACITY - 1: writeBuffers[wi].put(position, (byte)size); if (wi == lastIndex) { wi = 0; } else { wi++; } position = 0; writeBuffers[wi].put(position++, (byte)(size >> 8)); writeBuffers[wi].put(position++, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); writeBuffers[wi].put(position++, (byte)hashcode); writeBuffers[wi].put(position++, (byte)(hashcode >> 8)); writeBuffers[wi].put(position++, (byte)(hashcode >> 16)); writeBuffers[wi].put(position, (byte)(hashcode >> 24)); break; case BUFFER_CAPACITY - 2: writeBuffers[wi].put(position++, (byte)size); writeBuffers[wi].put(position, (byte)(size >> 8)); if (wi == lastIndex) { wi = 0; } else { wi++; } position = 0; writeBuffers[wi].put(position++, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); writeBuffers[wi].put(position++, (byte)hashcode); writeBuffers[wi].put(position++, (byte)(hashcode >> 8)); writeBuffers[wi].put(position++, (byte)(hashcode >> 16)); writeBuffers[wi].put(position, (byte)(hashcode >> 24)); break; case BUFFER_CAPACITY - 3: writeBuffers[wi].put(position++, (byte)size); writeBuffers[wi].put(position++, (byte)(size >> 8)); writeBuffers[wi].put(position, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); if (wi == lastIndex) { wi = 0; } else { wi++; } position = 0; writeBuffers[wi].put(position++, (byte)hashcode); writeBuffers[wi].put(position++, (byte)(hashcode >> 8)); writeBuffers[wi].put(position++, (byte)(hashcode >> 16)); writeBuffers[wi].put(position, (byte)(hashcode >> 24)); break; case BUFFER_CAPACITY - 4: writeBuffers[wi].put(position++, (byte)size); writeBuffers[wi].put(position++, (byte)(size >> 8)); writeBuffers[wi].put(position++, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); writeBuffers[wi].put(position, (byte)hashcode); if (wi == lastIndex) { wi = 0; } else { wi++; } position = 0; writeBuffers[wi].put(position++, (byte)(hashcode >> 8)); writeBuffers[wi].put(position++, (byte)(hashcode >> 16)); writeBuffers[wi].put(position, (byte)(hashcode >> 24)); break; case BUFFER_CAPACITY - 5: writeBuffers[wi].put(position++, (byte)size); writeBuffers[wi].put(position++, (byte)(size >> 8)); writeBuffers[wi].put(position++, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); writeBuffers[wi].put(position++, (byte)hashcode); writeBuffers[wi].put(position, (byte)(hashcode >> 8)); if (wi == lastIndex) { wi = 0; } else { wi++; } position = 0; writeBuffers[wi].put(position++, (byte)(hashcode >> 16)); writeBuffers[wi].put(position, (byte)(hashcode >> 24)); break; case BUFFER_CAPACITY - 6: writeBuffers[wi].put(position++, (byte)size); writeBuffers[wi].put(position++, (byte)(size >> 8)); writeBuffers[wi].put(position++, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); writeBuffers[wi].put(position++, (byte)hashcode); writeBuffers[wi].put(position++, (byte)(hashcode >> 8)); writeBuffers[wi].put(position, (byte)(hashcode >> 16)); if (wi == lastIndex) { wi = 0; } else { wi++; } position = 0; writeBuffers[wi].put(position, (byte)(hashcode >> 24)); break; default: writeBuffers[wi].put(position++, (byte)size); writeBuffers[wi].put(position++, (byte)(size >> 8)); writeBuffers[wi].put(position++, com.datatorrent.bufferserver.packet.MessageType.PAYLOAD_VALUE); writeBuffers[wi].put(position++, (byte)hashcode); writeBuffers[wi].put(position++, (byte)(hashcode >> 8)); writeBuffers[wi].put(position++, (byte)(hashcode >> 16)); writeBuffers[wi].put(position, (byte)(hashcode >> 24)); break; } synchronized (readBuffers[index]) { readBuffers[index].limit(BUFFER_CAPACITY); } } } if (!write) { key.interestOps(key.interestOps() | SelectionKey.OP_WRITE); write = true; key.selector().wakeup(); } } FastPublisher(String id, int streamingWindowThroughput); @Override void read(); @Override void write(); @Override void handleException(Exception cce, EventLoop el); @Override void registered(SelectionKey key); @Override void unregistered(SelectionKey key); @Override void setup(StreamContext context); @Override void teardown(); @Override void activate(StreamContext context); @Override void deactivate(); @Override @SuppressWarnings("SleepWhileInLoop") void put(Object tuple); @Override boolean putControl(ControlTuple payload); @SuppressWarnings("SleepWhileInLoop") void advanceWriteBuffer(); @Override int getCount(boolean reset); @Override void connected(); @Override void disconnected(); static final int BUFFER_CAPACITY; }
@Test public void testSerialization() throws Exception { FastPublisherImpl publisher = new FastPublisherImpl(24 * 1024); final String message = "hello!"; publisher.put(message); byte[] buffer = publisher.consume(); FastSubscriber subscriber = new FastSubscriber("subscriber", 1024); subscriber.serde = subscriber.statefulSerde = new DefaultStatefulStreamCodec<>(); SweepableReservoir sr = subscriber.acquireReservoir("res", 1024); sr.setSink(new Sink<Object>() { @Override public void put(Object tuple) { assert (tuple.equals(message)); } @Override public int getCount(boolean reset) { throw new UnsupportedOperationException("Not supported yet."); } }); int size = buffer[0]; size |= buffer[1] << 8; Assert.assertEquals("size", buffer.length - 2, size); subscriber.onMessage(buffer, 2, buffer.length - 2); sr.sweep(); sr.sweep(); for (int i = 0; i < 1024; i++) { publisher.put(message); } buffer = publisher.consume(); assert (buffer.length == (size + 2) * 1024); int index = 0; for (int i = 0; i < 1024; i++) { size = buffer[index++]; size |= buffer[index++] << 8; subscriber.onMessage(buffer, index, size); index += size; } sr.sweep(); sr.sweep(); for (int i = 0; i < 1024; i++) { publisher.put(message); } buffer = publisher.consume(); assert (buffer.length == (size + 2) * 1024); index = 0; for (int i = 0; i < 1024; i++) { size = buffer[index++]; size |= buffer[index++] << 8; subscriber.onMessage(buffer, index, size); index += size; } sr.sweep(); sr.sweep(); }