// Path: identity-service/src/main/java/eu/nimble/core/infrastructure/identity/config/KafkaConfig.java
// @Configuration
// @EnableKafka
// public class KafkaConfig {
// 
//     @Value("${spring.kafka.bootstrap-servers}")
//     private String bootstrapServers;
// 
//     @Value("${spring.kafka.properties.sasl.jaas.config}")
//     private String jaasConfig;
// 
//     @Value("${spring.kafka.consumer.group-id}")
//     private String consumerGroupId;
// 
//     @Bean
//     public Map<String, Object> producerConfigs() {
//         Map<String, Object> props = new HashMap<>();
//         props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//         if (StringUtils.isEmpty(jaasConfig) == false) {
//             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
//             props.put("sasl.mechanism", "PLAIN");
//             props.put("sasl.jaas.config", jaasConfig);
//         }
//         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
//         return props;
//     }
// 
//     @Bean
//     public ProducerFactory<String, AuthorizedCompanyUpdate> producerFactory() {
//         return new DefaultKafkaProducerFactory<>(producerConfigs());
//     }
// 
//     @Bean
//     public KafkaTemplate<String, AuthorizedCompanyUpdate> kafkaTemplate() {
//         return new KafkaTemplate<>(producerFactory());
//     }
// 
//     @Bean
//     public Map<String, Object> consumerConfigs() {
//         Map<String, Object> props = new HashMap<>();
//         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//         if (StringUtils.isEmpty(jaasConfig) == false) {
//             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
//             props.put("sasl.mechanism", "PLAIN");
//             props.put("sasl.jaas.config", jaasConfig);
//         }
//         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//         props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
//         return props;
//     }
// 
//     @Bean
//     public ConsumerFactory<String, AuthorizedCompanyUpdate> companyUpdateConsumerFactory() {
//         return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), new JsonDeserializer<>(AuthorizedCompanyUpdate.class));
//     }
// 
//     @Bean
//     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, AuthorizedCompanyUpdate>> companyUpdatesKafkaListenerContainerFactory() {
//         ConcurrentKafkaListenerContainerFactory<String, AuthorizedCompanyUpdate> factory = new ConcurrentKafkaListenerContainerFactory<>();
//         factory.setConsumerFactory(companyUpdateConsumerFactory());
//         return factory;
//     }
// 
//     public static class AuthorizedCompanyUpdate {
//         private String companyId;
//         private String accessToken;
// 
//         public AuthorizedCompanyUpdate() {
//         }
// 
//         public AuthorizedCompanyUpdate(String companyId, String accessToken) {
//             this.companyId = companyId;
//             this.accessToken = accessToken;
//         }
// 
//         public String getCompanyId() {
//             return companyId;
//         }
// 
//         public void setCompanyId(String companyID) {
//             this.companyId = companyID;
//         }
// 
//         public String getAccessToken() {
//             return accessToken;
//         }
// 
//         public void setAccessToken(String accessToken) {
//             this.accessToken = accessToken;
//         }
//     }
// }


import eu.nimble.core.infrastructure.identity.config.KafkaConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
package eu.nimble.core.infrastructure.identity.messaging;


/**
 * Created by Johannes Innerbichler on 27.09.18.
 */
@Component
public class KafkaReceiver {
    @KafkaListener(topics = "${nimble.kafka.topics.companyUpdates}", containerFactory = "companyUpdatesKafkaListenerContainerFactory")
// Path: identity-service/src/main/java/eu/nimble/core/infrastructure/identity/config/KafkaConfig.java
// @Configuration
// @EnableKafka
// public class KafkaConfig {
// 
//     @Value("${spring.kafka.bootstrap-servers}")
//     private String bootstrapServers;
// 
//     @Value("${spring.kafka.properties.sasl.jaas.config}")
//     private String jaasConfig;
// 
//     @Value("${spring.kafka.consumer.group-id}")
//     private String consumerGroupId;
// 
//     @Bean
//     public Map<String, Object> producerConfigs() {
//         Map<String, Object> props = new HashMap<>();
//         props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//         if (StringUtils.isEmpty(jaasConfig) == false) {
//             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
//             props.put("sasl.mechanism", "PLAIN");
//             props.put("sasl.jaas.config", jaasConfig);
//         }
//         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
//         return props;
//     }
// 
//     @Bean
//     public ProducerFactory<String, AuthorizedCompanyUpdate> producerFactory() {
//         return new DefaultKafkaProducerFactory<>(producerConfigs());
//     }
// 
//     @Bean
//     public KafkaTemplate<String, AuthorizedCompanyUpdate> kafkaTemplate() {
//         return new KafkaTemplate<>(producerFactory());
//     }
// 
//     @Bean
//     public Map<String, Object> consumerConfigs() {
//         Map<String, Object> props = new HashMap<>();
//         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//         if (StringUtils.isEmpty(jaasConfig) == false) {
//             props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
//             props.put("sasl.mechanism", "PLAIN");
//             props.put("sasl.jaas.config", jaasConfig);
//         }
//         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//         props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroupId);
//         return props;
//     }
// 
//     @Bean
//     public ConsumerFactory<String, AuthorizedCompanyUpdate> companyUpdateConsumerFactory() {
//         return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new StringDeserializer(), new JsonDeserializer<>(AuthorizedCompanyUpdate.class));
//     }
// 
//     @Bean
//     public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, AuthorizedCompanyUpdate>> companyUpdatesKafkaListenerContainerFactory() {
//         ConcurrentKafkaListenerContainerFactory<String, AuthorizedCompanyUpdate> factory = new ConcurrentKafkaListenerContainerFactory<>();
//         factory.setConsumerFactory(companyUpdateConsumerFactory());
//         return factory;
//     }
// 
//     public static class AuthorizedCompanyUpdate {
//         private String companyId;
//         private String accessToken;
// 
//         public AuthorizedCompanyUpdate() {
//         }
// 
//         public AuthorizedCompanyUpdate(String companyId, String accessToken) {
//             this.companyId = companyId;
//             this.accessToken = accessToken;
//         }
// 
//         public String getCompanyId() {
//             return companyId;
//         }
// 
//         public void setCompanyId(String companyID) {
//             this.companyId = companyID;
//         }
// 
//         public String getAccessToken() {
//             return accessToken;
//         }
// 
//         public void setAccessToken(String accessToken) {
//             this.accessToken = accessToken;
//         }
//     }
// }


// Path: identity-service/src/main/java/eu/nimble/core/infrastructure/identity/messaging/KafkaReceiver.java
import eu.nimble.core.infrastructure.identity.config.KafkaConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

package eu.nimble.core.infrastructure.identity.messaging;


/**
 * Created by Johannes Innerbichler on 27.09.18.
 */
@Component
public class KafkaReceiver {
    @KafkaListener(topics = "${nimble.kafka.topics.companyUpdates}", containerFactory = "companyUpdatesKafkaListenerContainerFactory")
        public void receiveCompanyUpdates(ConsumerRecord<String, KafkaConfig.AuthorizedCompanyUpdate> consumerRecord) {