package net.wasdev.gameon.room;

import java.util.Properties;
import javax.enterprise.context.ApplicationScoped;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import org.apache.kafka.clients.producer.*;

@ApplicationScoped
public class Kafka {

  protected String kafkaUrl;

   private Producer<String,String> producer=null;

   public Kafka(){
     getConfig();
     initProducer();
   }

   private void getConfig() {
       try {
           kafkaUrl = (String) new InitialContext().lookup("kafkaUrl");
       } catch (NamingException e) {
       }
       if (kafkaUrl == null ) {
           throw new IllegalStateException("kafkaUrl("+String.valueOf(kafkaUrl)+") was not found, check server.xml/server.env");
       }
   }

   private void initProducer(){
       System.out.println("Initializing kafka producer for url "+kafkaUrl);
       Properties producerProps = new Properties();
       producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
       producerProps.put("acks","all");
       producerProps.put("retries",0);
       producerProps.put("batch.size",16384);
       producerProps.put("zookeeper.session.timeout.ms",1000);
       producerProps.put("linger.ms",1);
       producerProps.put("buffer.memory",33554432);
       producerProps.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
       producerProps.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
       producer = new KafkaProducer<String, String>(producerProps);
   }

   public void publishMessage(String topic, String key, String message){
     System.out.println("Publishing to kafka, creating record");
     ProducerRecord<String,String> pr = new ProducerRecord<String,String>(topic, key, message);
     System.out.println("Publishing to kafka, sending record");
     producer.send(pr);
     System.out.println("Publishing to kafka, sent record");
   }
}

@@ -1021,11 +1021,63 @@
 s%22,%22
-all
+-1%22);%0A       producerProps.put(%22client.id%22,%22gameon-room
 %22);%0A
@@ -1410,32 +1410,32 @@
 ngSerializer%22);%0A
-
        producerP
@@ -1512,32 +1512,896 @@
 ingSerializer%22);
+%0A%0A       //this is a cheat, we need to enable ssl when talking to message hub, and not to kafka locally%0A       //the easiest way to know which we are running on, is to check how many hosts are in kafkaUrl%0A       //locally for kafka there'll only ever be one, and messagehub gives us a whole bunch..%0A       boolean multipleHosts = kafkaUrl.indexOf(%22,%22) != -1;%0A       if(multipleHosts)%7B%0A         producerProps.put(%22security.protocol%22,%22SASL_SSL%22);%0A         producerProps.put(%22ssl.protocol%22,%22TLSv1.2%22);%0A         producerProps.put(%22ssl.enabled.protocols%22,%22TLSv1.2%22);%0A         producerProps.put(%22ssl.truststore.location%22,%22/opt/ibm/java/jre/security/cacerts%22);%0A         producerProps.put(%22ssl.truststore.password%22,%22changeit%22);%0A         producerProps.put(%22ssl.truststore.type%22,%22JKS%22);%0A         producerProps.put(%22ssl.endpoint.identification.algorithm%22,%22HTTPS%22);%0A       %7D%0A%0A
 %0A       producer

Add ssl props based on messagehub sample