@Test public void testBadNotification() { decoder = new TrackMessageSizeDecoder(DELIMITER, false); String message = "38|{\n \"id\" : \"G1J__clURxSyAgLX8O5zgw\"\n}"; List<String> expected = new ArrayList<String>() { { add("{\n \"id\" : \"G1J__clURxSyAgLX8O5zgw\"\n}"); } }; Decoded<List<String>> result = decoder.decode(Event.MESSAGE, message); assertEquals(result.decoded(), expected); }
@Override public Decoded<List<String>> decode(Event type, String message) { if (type.equals(Event.MESSAGE)) { if (skipFirstMessage.getAndSet(false)) return empty; Decoded<List<String>> messages = new Decoded<List<String>>(new LinkedList<String>()); message = messagesBuffer.append(message).toString().replace(delimiter, charReplacement); messagesBuffer.setLength(0); if (message.indexOf(charReplacement) != -1) { String[] tokens = message.split(charReplacement); int pos = 1; int length = Integer.valueOf(tokens[0]); String m; while (pos < tokens.length) { m = tokens[pos]; if (m.length() >= length) { messages.decoded().add(m.substring(0, length)); String t = m.substring(length); if (t.length() > 0) { length = Integer.valueOf(t); } } else { if (pos != 1) { messagesBuffer.append(length).append(delimiter).append(m); } else { messagesBuffer.append(message); } if (messages.decoded().size() > 0) { return messages; } else { return new Decoded<List<String>>(new LinkedList<String>(), Decoded.ACTION.ABORT); } } pos++; } } return messages; } else { return empty; } }
TrackMessageSizeDecoder implements ReplayDecoder<String, String> { @Override public Decoded<List<String>> decode(Event type, String message) { if (type.equals(Event.MESSAGE)) { if (skipFirstMessage.getAndSet(false)) return empty; Decoded<List<String>> messages = new Decoded<List<String>>(new LinkedList<String>()); message = messagesBuffer.append(message).toString().replace(delimiter, charReplacement); messagesBuffer.setLength(0); if (message.indexOf(charReplacement) != -1) { String[] tokens = message.split(charReplacement); int pos = 1; int length = Integer.valueOf(tokens[0]); String m; while (pos < tokens.length) { m = tokens[pos]; if (m.length() >= length) { messages.decoded().add(m.substring(0, length)); String t = m.substring(length); if (t.length() > 0) { length = Integer.valueOf(t); } } else { if (pos != 1) { messagesBuffer.append(length).append(delimiter).append(m); } else { messagesBuffer.append(message); } if (messages.decoded().size() > 0) { return messages; } else { return new Decoded<List<String>>(new LinkedList<String>(), Decoded.ACTION.ABORT); } } pos++; } } return messages; } else { return empty; } } }
TrackMessageSizeDecoder implements ReplayDecoder<String, String> { @Override public Decoded<List<String>> decode(Event type, String message) { if (type.equals(Event.MESSAGE)) { if (skipFirstMessage.getAndSet(false)) return empty; Decoded<List<String>> messages = new Decoded<List<String>>(new LinkedList<String>()); message = messagesBuffer.append(message).toString().replace(delimiter, charReplacement); messagesBuffer.setLength(0); if (message.indexOf(charReplacement) != -1) { String[] tokens = message.split(charReplacement); int pos = 1; int length = Integer.valueOf(tokens[0]); String m; while (pos < tokens.length) { m = tokens[pos]; if (m.length() >= length) { messages.decoded().add(m.substring(0, length)); String t = m.substring(length); if (t.length() > 0) { length = Integer.valueOf(t); } } else { if (pos != 1) { messagesBuffer.append(length).append(delimiter).append(m); } else { messagesBuffer.append(message); } if (messages.decoded().size() > 0) { return messages; } else { return new Decoded<List<String>>(new LinkedList<String>(), Decoded.ACTION.ABORT); } } pos++; } } return messages; } else { return empty; } } TrackMessageSizeDecoder();  TrackMessageSizeDecoder(boolean protocolEnabled);  TrackMessageSizeDecoder(String delimiter, boolean protocolEnabled); }
TrackMessageSizeDecoder implements ReplayDecoder<String, String> { @Override public Decoded<List<String>> decode(Event type, String message) { if (type.equals(Event.MESSAGE)) { if (skipFirstMessage.getAndSet(false)) return empty; Decoded<List<String>> messages = new Decoded<List<String>>(new LinkedList<String>()); message = messagesBuffer.append(message).toString().replace(delimiter, charReplacement); messagesBuffer.setLength(0); if (message.indexOf(charReplacement) != -1) { String[] tokens = message.split(charReplacement); int pos = 1; int length = Integer.valueOf(tokens[0]); String m; while (pos < tokens.length) { m = tokens[pos]; if (m.length() >= length) { messages.decoded().add(m.substring(0, length)); String t = m.substring(length); if (t.length() > 0) { length = Integer.valueOf(t); } } else { if (pos != 1) { messagesBuffer.append(length).append(delimiter).append(m); } else { messagesBuffer.append(message); } if (messages.decoded().size() > 0) { return messages; } else { return new Decoded<List<String>>(new LinkedList<String>(), Decoded.ACTION.ABORT); } } pos++; } } return messages; } else { return empty; } } TrackMessageSizeDecoder();  TrackMessageSizeDecoder(boolean protocolEnabled);  TrackMessageSizeDecoder(String delimiter, boolean protocolEnabled); @Override Decoded<List<String>> decode(Event type, String message); }
TrackMessageSizeDecoder implements ReplayDecoder<String, String> { @Override public Decoded<List<String>> decode(Event type, String message) { if (type.equals(Event.MESSAGE)) { if (skipFirstMessage.getAndSet(false)) return empty; Decoded<List<String>> messages = new Decoded<List<String>>(new LinkedList<String>()); message = messagesBuffer.append(message).toString().replace(delimiter, charReplacement); messagesBuffer.setLength(0); if (message.indexOf(charReplacement) != -1) { String[] tokens = message.split(charReplacement); int pos = 1; int length = Integer.valueOf(tokens[0]); String m; while (pos < tokens.length) { m = tokens[pos]; if (m.length() >= length) { messages.decoded().add(m.substring(0, length)); String t = m.substring(length); if (t.length() > 0) { length = Integer.valueOf(t); } } else { if (pos != 1) { messagesBuffer.append(length).append(delimiter).append(m); } else { messagesBuffer.append(message); } if (messages.decoded().size() > 0) { return messages; } else { return new Decoded<List<String>>(new LinkedList<String>(), Decoded.ACTION.ABORT); } } pos++; } } return messages; } else { return empty; } } TrackMessageSizeDecoder();  TrackMessageSizeDecoder(boolean protocolEnabled);  TrackMessageSizeDecoder(String delimiter, boolean protocolEnabled); @Override Decoded<List<String>> decode(Event type, String message);  }