MessageHolderWithTokensKafkaProducer.java

public class MessageHolderWithTokensKafkaProducer {

    @Producer
    private SimpleKafkaProducer<String, String> producer;

    private final String ADM_TOPIC = "AdmToken_Topic";

    private final String APNS_TOPIC = "APNsToken_Topic";

    private final String FCM_TOPIC = "FCMToken_Topic";

    private final String MPNS_TOPIC = "MPNSToken_Topic";

    private final String MOZ_TOPIC = "SimplePushToken_Topic";

    private final String WNS_TOPIC = "WNSToken_Topic";

    public void queueMessageVariantForProcessing(@Observes @DispatchToQueue MessageHolderWithTokens msg) {

        final String pushTopic = selectTopic(msg.getVariant().getType());
        producer.send(pushTopic, msg);
    }

    private String selectTopic(final VariantType variantType) {
        switch (variantType) {
            case ADM:
                return ADM_TOPIC;
            case ANDROID:
                return FCM_TOPIC;
            case IOS:
                return APNS_TOPIC;
            case SIMPLE_PUSH:
                return MOZ_TOPIC;
            case WINDOWS_MPNS:
                return MPNS_TOPIC;
            case WINDOWS_WNS:
                return WNS_TOPIC;
            default:
                throw new IllegalStateException("Unknown variant type queue");
        }
    }