@Test public void shouldBeBackwardCompatible() throws Exception { UUID processId = UUID.randomUUID(); Set<TaskId> activeTasks = new HashSet<>(Arrays.asList(new TaskId(0, 0), new TaskId(0, 1), new TaskId(1, 0))); Set<TaskId> standbyTasks = new HashSet<>(Arrays.asList(new TaskId(1, 1), new TaskId(2, 0))); final ByteBuffer v1Encoding = encodePreviousVersion(processId, activeTasks, standbyTasks); final SubscriptionInfo decode = SubscriptionInfo.decode(v1Encoding); assertEquals(activeTasks, decode.prevTasks); assertEquals(standbyTasks, decode.standbyTasks); assertEquals(processId, decode.processId); assertNull(decode.userEndPoint); }
public static SubscriptionInfo decode(ByteBuffer data) { data.rewind(); int version = data.getInt(); if (version == CURRENT_VERSION || version == 1) { UUID processId = new UUID(data.getLong(), data.getLong()); Set<TaskId> prevTasks = new HashSet<>(); int numPrevs = data.getInt(); for (int i = 0; i < numPrevs; i++) { TaskId id = TaskId.readFrom(data); prevTasks.add(id); } Set<TaskId> standbyTasks = new HashSet<>(); int numCached = data.getInt(); for (int i = 0; i < numCached; i++) { standbyTasks.add(TaskId.readFrom(data)); } String userEndPoint = null; if (version == CURRENT_VERSION) { int bytesLength = data.getInt(); if (bytesLength != 0) { byte[] bytes = new byte[bytesLength]; data.get(bytes); userEndPoint = new String(bytes, Charset.forName("UTF-8")); } } return new SubscriptionInfo(version, processId, prevTasks, standbyTasks, userEndPoint); } else { TaskAssignmentException ex = new TaskAssignmentException("unable to decode subscription data: version=" + version); log.error(ex.getMessage(), ex); throw ex; } }
SubscriptionInfo { public static SubscriptionInfo decode(ByteBuffer data) { data.rewind(); int version = data.getInt(); if (version == CURRENT_VERSION || version == 1) { UUID processId = new UUID(data.getLong(), data.getLong()); Set<TaskId> prevTasks = new HashSet<>(); int numPrevs = data.getInt(); for (int i = 0; i < numPrevs; i++) { TaskId id = TaskId.readFrom(data); prevTasks.add(id); } Set<TaskId> standbyTasks = new HashSet<>(); int numCached = data.getInt(); for (int i = 0; i < numCached; i++) { standbyTasks.add(TaskId.readFrom(data)); } String userEndPoint = null; if (version == CURRENT_VERSION) { int bytesLength = data.getInt(); if (bytesLength != 0) { byte[] bytes = new byte[bytesLength]; data.get(bytes); userEndPoint = new String(bytes, Charset.forName("UTF-8")); } } return new SubscriptionInfo(version, processId, prevTasks, standbyTasks, userEndPoint); } else { TaskAssignmentException ex = new TaskAssignmentException("unable to decode subscription data: version=" + version); log.error(ex.getMessage(), ex); throw ex; } } }
SubscriptionInfo { public static SubscriptionInfo decode(ByteBuffer data) { data.rewind(); int version = data.getInt(); if (version == CURRENT_VERSION || version == 1) { UUID processId = new UUID(data.getLong(), data.getLong()); Set<TaskId> prevTasks = new HashSet<>(); int numPrevs = data.getInt(); for (int i = 0; i < numPrevs; i++) { TaskId id = TaskId.readFrom(data); prevTasks.add(id); } Set<TaskId> standbyTasks = new HashSet<>(); int numCached = data.getInt(); for (int i = 0; i < numCached; i++) { standbyTasks.add(TaskId.readFrom(data)); } String userEndPoint = null; if (version == CURRENT_VERSION) { int bytesLength = data.getInt(); if (bytesLength != 0) { byte[] bytes = new byte[bytesLength]; data.get(bytes); userEndPoint = new String(bytes, Charset.forName("UTF-8")); } } return new SubscriptionInfo(version, processId, prevTasks, standbyTasks, userEndPoint); } else { TaskAssignmentException ex = new TaskAssignmentException("unable to decode subscription data: version=" + version); log.error(ex.getMessage(), ex); throw ex; } } SubscriptionInfo(UUID processId, Set<TaskId> prevTasks, Set<TaskId> standbyTasks, String userEndPoint); private  SubscriptionInfo(int version, UUID processId, Set<TaskId> prevTasks, Set<TaskId> standbyTasks, String userEndPoint); }
SubscriptionInfo { public static SubscriptionInfo decode(ByteBuffer data) { data.rewind(); int version = data.getInt(); if (version == CURRENT_VERSION || version == 1) { UUID processId = new UUID(data.getLong(), data.getLong()); Set<TaskId> prevTasks = new HashSet<>(); int numPrevs = data.getInt(); for (int i = 0; i < numPrevs; i++) { TaskId id = TaskId.readFrom(data); prevTasks.add(id); } Set<TaskId> standbyTasks = new HashSet<>(); int numCached = data.getInt(); for (int i = 0; i < numCached; i++) { standbyTasks.add(TaskId.readFrom(data)); } String userEndPoint = null; if (version == CURRENT_VERSION) { int bytesLength = data.getInt(); if (bytesLength != 0) { byte[] bytes = new byte[bytesLength]; data.get(bytes); userEndPoint = new String(bytes, Charset.forName("UTF-8")); } } return new SubscriptionInfo(version, processId, prevTasks, standbyTasks, userEndPoint); } else { TaskAssignmentException ex = new TaskAssignmentException("unable to decode subscription data: version=" + version); log.error(ex.getMessage(), ex); throw ex; } } SubscriptionInfo(UUID processId, Set<TaskId> prevTasks, Set<TaskId> standbyTasks, String userEndPoint); private  SubscriptionInfo(int version, UUID processId, Set<TaskId> prevTasks, Set<TaskId> standbyTasks, String userEndPoint); ByteBuffer encode(); static SubscriptionInfo decode(ByteBuffer data); @Override int hashCode(); @Override boolean equals(Object o); }
SubscriptionInfo { public static SubscriptionInfo decode(ByteBuffer data) { data.rewind(); int version = data.getInt(); if (version == CURRENT_VERSION || version == 1) { UUID processId = new UUID(data.getLong(), data.getLong()); Set<TaskId> prevTasks = new HashSet<>(); int numPrevs = data.getInt(); for (int i = 0; i < numPrevs; i++) { TaskId id = TaskId.readFrom(data); prevTasks.add(id); } Set<TaskId> standbyTasks = new HashSet<>(); int numCached = data.getInt(); for (int i = 0; i < numCached; i++) { standbyTasks.add(TaskId.readFrom(data)); } String userEndPoint = null; if (version == CURRENT_VERSION) { int bytesLength = data.getInt(); if (bytesLength != 0) { byte[] bytes = new byte[bytesLength]; data.get(bytes); userEndPoint = new String(bytes, Charset.forName("UTF-8")); } } return new SubscriptionInfo(version, processId, prevTasks, standbyTasks, userEndPoint); } else { TaskAssignmentException ex = new TaskAssignmentException("unable to decode subscription data: version=" + version); log.error(ex.getMessage(), ex); throw ex; } } SubscriptionInfo(UUID processId, Set<TaskId> prevTasks, Set<TaskId> standbyTasks, String userEndPoint); private  SubscriptionInfo(int version, UUID processId, Set<TaskId> prevTasks, Set<TaskId> standbyTasks, String userEndPoint); ByteBuffer encode(); static SubscriptionInfo decode(ByteBuffer data); @Override int hashCode(); @Override boolean equals(Object o); final int version; final UUID processId; final Set<TaskId> prevTasks; final Set<TaskId> standbyTasks; final String userEndPoint; }