+ BytesAndType serializeAsClientMessage(Integer key, Long value)
+ {
+ ClientMessage message = new ClientMessage();
+ message.setClient(key.toString());
+ message.setMessage(value.toString());
+ return new BytesAndType(serialize(message), "message");
+ }
+
+ BytesAndType serializeAsGreeting(Integer key)
+ {
+ Greeting greeting = new Greeting();
+ greeting.setName(key.toString());
+ greeting.setWhen(LocalDateTime.now());
+ return new BytesAndType(serialize(greeting), "greeting");
+ }
+
+ BytesAndType serializeString(String message, String messageType)
+ {
+ return new BytesAndType(new Bytes(message.getBytes()), messageType);
+ }
+
+ Bytes serialize(Object message)
+ {
+ return new Bytes(valueSerializer.serialize(TOPIC, message));
+ }
+