PublishMBMessagesToKafka.java

@Component(
	immediate = true,
	service = ModelListener.class
)
public class MBMessagesModelListener extends BaseModelListener<MBMessage> {

	@Override
	public void onAfterCreate(MBMessage model)
		throws ModelListenerException {

		super.onAfterUpdate(model);

		publishMessage(model);
	}

	@Override
	public void onAfterUpdate(MBMessage model) throws ModelListenerException {
		super.onAfterUpdate(model);

		publishMessage(model);
	}

	protected Message buildMessage(MBMessage mbMessage)  {
		Message message = new Message();

		message.put("topic", _ENTITY_EVENT);
		message.setPayload(buildPayload(mbMessage));

		return message;
	}

	protected String buildPayload(MBMessage mbMessage) {
		StringBundler sb = new StringBundler(6);

		sb.append(mbMessage.getCompanyId());
		sb.append(StringPool.SEMICOLON);
		sb.append(mbMessage.getUserId());
		sb.append(StringPool.SEMICOLON);
		sb.append(mbMessage.getCreateDate());
		sb.append(StringPool.SEMICOLON);
		sb.append(mbMessage.getMessageId());
		sb.append(StringPool.SEMICOLON);
		sb.append(mbMessage.getSubject());
		sb.append(StringPool.SEMICOLON);
		sb.append(mbMessage.getBody());
		sb.append(StringPool.SEMICOLON);

		return sb.toString();
	}

	protected void publishMessage(MBMessage mbMessage) {
		_messageBus.sendMessage(
			"kafka_destination", buildMessage(mbMessage));
	}

	@Reference
	protected void setMessageBus(MessageBus messageBus) {
		_messageBus = messageBus;
	}

	protected void unsetMessageBus(MessageBus messageBus) {
		_messageBus = null;
	}

	private static final String _ENTITY_EVENT = "mb_messages";

	private MessageBus _messageBus;

}