SubscribeRequestTuple extends RequestTuple { public static byte[] getSerializedRequest(final String version, final String id, final String down_type, final String upstream_id, final int mask, final Collection<Integer> partitions, final long startingWindowId, final int bufferSize) { byte[] array = new byte[4096]; int offset = 0; array[offset++] = MessageType.SUBSCRIBER_REQUEST_VALUE; offset = Tuple.writeString(version == null ? CLASSIC_VERSION : version, array, offset); offset = Tuple.writeString(id, array, offset); int baseSeconds = (int)(startingWindowId >> 32); offset = VarInt.write(baseSeconds, array, offset); int windowId = (int)startingWindowId; offset = VarInt.write(windowId, array, offset); offset = Tuple.writeString(down_type, array, offset); offset = Tuple.writeString(upstream_id, array, offset); if (partitions == null || partitions.isEmpty()) { offset = VarInt.write(0, array, offset); } else { offset = VarInt.write(partitions.size(), array, offset); offset = VarInt.write(mask, array, offset); for (int i : partitions) { offset = VarInt.write(i, array, offset); } } offset = VarInt.write(bufferSize, array, offset); return Arrays.copyOfRange(array, 0, offset); } SubscribeRequestTuple(byte[] array, int offset, int length); @Override MessageType getType(); @Override void parse(); boolean isParsed(); String getStreamType(); @Override int getWindowId(); @Override int getBaseSeconds(); @Override String getVersion(); @Override String getIdentifier(); String getUpstreamIdentifier(); int getMask(); @SuppressWarnings(value = "ReturnOfCollectionOrArrayField") int[] getPartitions(); int getBufferSize(); static byte[] getSerializedRequest(final String version, final String id, final String down_type,
      final String upstream_id, final int mask, final Collection<Integer> partitions, final long startingWindowId,
      final int bufferSize); @Override String toString(); static final String EMPTY_STRING; }
@Test public void testGetSerializedRequest() { String id = "SubscriberId"; String down_type = "SubscriberId/StreamType"; String upstream_id = "PublisherId"; int mask = 7; ArrayList<Integer> partitions = new ArrayList<>(); partitions.add(5); long startingWindowId = 0xcafebabe00000078L; byte[] serial = getSerializedRequest(null, id, down_type, upstream_id, mask, partitions, startingWindowId, 32 * 1024); SubscribeRequestTuple tuple = (SubscribeRequestTuple)Tuple.getTuple(serial, 0, serial.length); assertEquals(tuple.getIdentifier(), id, "Identifier"); assertEquals(tuple.getStreamType(), down_type, "UpstreamType"); assertEquals(tuple.getUpstreamIdentifier(), upstream_id, "UpstreamId"); assertEquals(tuple.getMask(), mask, "Mask"); assertEquals(tuple.getBufferSize(), 32 * 1024, "BufferSize"); int[] parts = tuple.getPartitions(); assertNotNull(parts); assertEquals(parts.length, 1); assertEquals(parts[0], 5); assertEquals((long)tuple.getBaseSeconds() << 32 | tuple.getWindowId(), startingWindowId, "Window"); serial = getSerializedRequest(null, id, down_type, upstream_id, 0, null, startingWindowId, 32 * 1024); tuple = (SubscribeRequestTuple)Tuple.getTuple(serial, 0, serial.length); assertEquals(tuple.getIdentifier(), id, "Identifier"); assertEquals(tuple.getStreamType(), down_type, "UpstreamType"); assertEquals(tuple.getUpstreamIdentifier(), upstream_id, "UpstreamId"); assertEquals(tuple.getMask(), 0, "Mask"); assertEquals(tuple.getBufferSize(), 32 * 1024, "BufferSize"); assertNull(tuple.getPartitions()); assertEquals((long)tuple.getBaseSeconds() << 32 | tuple.getWindowId(), startingWindowId, "Window"); }