<|start_filename|>src/main/java/com/iniesta/flink/connector/mysql/MysqlDao.java<|end_filename|>
package com.iniesta.flink.connector.mysql;

import java.io.Serializable;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.iniesta.flink.connector.mqtt.MqttMessage;

public class MysqlDao implements Serializable {

	private static final Logger logger = LoggerFactory.getLogger(MysqlDao.class);

	private String url;
	private String user;
	private String pass;
	private String driver;
	private transient Connection connection;

	private final String defaultInsertQuery = "insert into messages(topic, payload, ts) values (?,?,now())";
	private final String defaultCreateQuery = "create table if not exists messages(id int not null auto_increment primary key, topic varchar(255), payload varchar(10000),ts datetime)";
	private String sqlCreate;
	private String sqlInsert;
	
	public MysqlDao(Properties props) {
		
		url = props.getProperty("jdbc.url");
		user = props.getProperty("jdbc.user");
		pass = props.getProperty("jdbc.pass");
		driver = props.getProperty("jdbc.driver");
		sqlCreate = props.getProperty("jdbc.sql.create", defaultCreateQuery);
		sqlInsert = props.getProperty("jdbc.sql.insert", defaultInsertQuery);
	}

	public void open() {
		if (connection == null) {
			try {
				Class.forName(driver);
				connection = DriverManager.getConnection(url, user, pass);
				createTable();
			} catch (ClassNotFoundException e) {
				e.printStackTrace();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
	}

	public void close() {
		if (connection != null) {
			logger.debug("Destroying connection to: {}:{}", url, user);
			try {
				connection.close();
			} catch (SQLException e) {
				e.printStackTrace();
			}
		}
		connection = null;
	}

	public void process(MqttMessage msg) {
		insertData(msg);
	}

	private static String escapeSQL(String s) {
		return s.replaceAll("'", "\\'");
	}
	
	private void insertData(MqttMessage msg) {
        try {
			PreparedStatement insertStmnt = connection.prepareStatement(sqlInsert );
            insertStmnt.setString(1, escapeSQL(msg.getTopic()));
            insertStmnt.setString(2, escapeSQL(msg.getPayload()));
            insertStmnt.execute();            
        } catch (SQLException e) {
            logger.error("Error during the updating of the event", e);
        }
    }

	private void createTable() {		
		try {
			Statement statement = connection.createStatement();
			statement.executeUpdate(sqlCreate);
		}catch(SQLException e) {
			logger.error("Error during the creation of the table", e);
		}
	}

	public List<String> testQuery() throws SQLException {
		List<String> output = new ArrayList<>();
		PreparedStatement ps = connection.prepareStatement("select id, a, b from data");
		ResultSet rs = ps.executeQuery();
		while(rs!=null && rs.next()) {
			String res = String.format("(%s,%s,%s)", rs.getInt(1), rs.getInt(2), rs.getInt(3));
			output.add(res);
		}
		return output;
	}
}



<|start_filename|>src/main/java/com/iniesta/flink/connector/mysql/MysqlSink.java<|end_filename|>
package com.iniesta.flink.connector.mysql;

import java.util.Properties;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import com.iniesta.flink.connector.mqtt.MqttMessage;

public class MysqlSink extends RichSinkFunction<MqttMessage> {

	private MysqlDao mysqlDao;

	public MysqlSink(Properties props) {
		mysqlDao = new MysqlDao(props);
	}

	@Override
	public void invoke(MqttMessage msg) throws Exception {
		mysqlDao.process(msg);
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		mysqlDao.open();
	}

	@Override
	public void close() throws Exception {
		super.close();
		mysqlDao.close();
	}

}


<|start_filename|>src/main/java/com/iniesta/flink/connector/mqtt/MqttMessage.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

import java.io.Serializable;

public class MqttMessage implements Serializable {

	private static final long serialVersionUID = -4673414704450588069L;
	
	private String topic;
	private String payload;

	public MqttMessage() {
	}

	public MqttMessage(String topic, String payload) {
		super();
		this.topic = topic;
		this.payload = payload;
	}

	public String getTopic() {
		return topic;
	}

	public void setTopic(String topic) {
		this.topic = topic;
	}

	public String getPayload() {
		return payload;
	}

	public void setPayload(String payload) {
		this.payload = payload;
	}

	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + ((payload == null) ? 0 : payload.hashCode());
		result = prime * result + ((topic == null) ? 0 : topic.hashCode());
		return result;
	}

	@Override
	public boolean equals(Object obj) {
		if (this == obj)
			return true;
		if (obj == null)
			return false;
		if (getClass() != obj.getClass())
			return false;
		MqttMessage other = (MqttMessage) obj;
		if (payload == null) {
			if (other.payload != null)
				return false;
		} else if (!payload.equals(other.payload))
			return false;
		if (topic == null) {
			if (other.topic != null)
				return false;
		} else if (!topic.equals(other.topic))
			return false;
		return true;
	}

	@Override
	public String toString() {
		return "MqttMessage [topic=" + topic + ", payload=" + payload + "]";
	}

}


<|start_filename|>src/main/java/com/iniesta/flink/connector/mqtt/MqttSource.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;

public class MqttSource extends RichSourceFunction<MqttMessage>{

	private String host;
	private int port;
	private String topic;
	private QoS qos;
	private boolean retain;

	public MqttSource(String host, String topic) {
		this(host, 1883, topic, QoS.AT_LEAST_ONCE, false);
	}
	
	public MqttSource(String host, int port, String topic) {
		this(host, port, topic, QoS.AT_LEAST_ONCE, false);
	}
	
	public MqttSource(String host, int port, String topic, QoS qos, boolean retain) {
		this.host = host;
		this.port = port;
		this.topic = topic;
		this.qos = qos;
		this.retain = retain;
	}
	
	@Override
	public void cancel() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void run(SourceContext<MqttMessage> sourceContext) throws Exception {
		System.out.println("[Source] Running source");
		MQTT mqtt = new MQTT();
		mqtt.setHost(host, port);
		BlockingConnection blockingConnection = mqtt.blockingConnection();
		blockingConnection.connect();
		System.out.println("[Source] Connected to mqtt broker");
		
		byte[] qoses = blockingConnection.subscribe(new Topic[] {new Topic(topic, qos)});
		System.out.println("[Source] Subscribed to " + topic);
		
		while(blockingConnection.isConnected()) {
			Message message = blockingConnection.receive();
			MqttMessage mmsg = new MqttMessage(message.getTopic(), new String(message.getPayload()));
			message.ack();
			sourceContext.collect(mmsg);
		}
		blockingConnection.disconnect();
	}

}


<|start_filename|>src/main/java/com/iniesta/com/SimpleSubscriber.java<|end_filename|>
package com.iniesta.com;

import java.io.FileInputStream;
import java.util.Properties;

import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;

import com.iniesta.flink.connector.mqtt.MqttMessage;

public class SimpleSubscriber {

	public static void main(String[] args) throws Exception {
		Properties props = new Properties();
		props.load(new FileInputStream(args[0]));
		
		String hostName = props.getProperty("mqtt.host");
		int port = Integer.parseInt(props.getProperty("mqtt.port", "1883"));		
		String topic = props.getProperty("mqtt.topic");
		
		
		MQTT mqtt = new MQTT();
		mqtt.setHost(hostName, port);
		BlockingConnection blockingConnection = mqtt.blockingConnection();
		try {
		blockingConnection.connect();
		} catch(Exception ex) {
			System.out.println("[Source] Not connected");
			ex.printStackTrace();
		}
		System.out.println("[Source] Connected to mqtt broker");
		
		byte[] qoses = blockingConnection.subscribe(new Topic[] {new Topic(topic, QoS.AT_LEAST_ONCE)});
		System.out.println("[Source] Subscribe to " + topic);
		
		while(blockingConnection.isConnected()) {
			Message message = blockingConnection.receive();
			System.out.println("[Source] Receiving message, now process internal message");
			MqttMessage mmsg = new MqttMessage(message.getTopic(), new String(message.getPayload()));
			System.out.println("[Source]: " + mmsg);
			message.ack();
			System.out.println("[msg] "+ mmsg);
		}
		blockingConnection.disconnect();
	}
}


<|start_filename|>src/main/java/com/iniesta/flink/connector/mqtt/SocketTextStreamWordCount2Mqtt.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class SocketTextStreamWordCount2Mqtt {

	//
	// Program
	//

	public static void main(String[] args) throws Exception {

		if (args.length != 2) {
			System.err.println("USAGE:\nSocketTextStreamWordCount <hostname> <port>");
			return;
		}

		String hostName = args[0];
		Integer port = Integer.parseInt(args[1]);

		// set up the execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// get input data
		DataStream<String> text = env.socketTextStream(hostName, port);

		DataStream<Tuple2<String, Integer>> counts =
				// split up the lines in pairs (2-tuples) containing: (word,1)
				text.flatMap(new LineSplitter())
						// group by the tuple field "0" and sum up tuple field "1"
						.keyBy(0).sum(1);

		// counts.print();
		counts.addSink(new MqttSink<>("cydonia", "/samples"));

		// execute program
		env.execute("Java WordCount from SocketTextStream Example");
	}

	//
	// User Functions
	//

	/**
	 * Implements the string tokenizer that splits sentences into words as a
	 * user-defined FlatMapFunction. The function takes a line (String) and splits
	 * it into multiple pairs in the form of "(word,1)" (Tuple2<String, Integer>).
	 */
	public static final class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> {

		@Override
		public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
			// normalize and split the line
			String[] tokens = value.toLowerCase().split("\\W+");

			// emit the pairs
			for (String token : tokens) {
				if (token.length() > 0) {
					out.collect(new Tuple2<String, Integer>(token, 1));
				}
			}
		}
	}
}


<|start_filename|>Dockerfile<|end_filename|>
FROM java:8

USER root

RUN mkdir /usr/mqtt2mysql

#RUN curl -Ls https://github.com/ajiniesta/flink-connector-mqtt/releases/download/0.1/flink-connector-mqtt-0.1.jar > /usr/mqtt2mysql/flink-con-mqtt.jar

COPY target/flink-connector-mqtt-*.jar /usr/mqtt2mysql/flink-con-mqtt.jar

CMD sleep 15 && java -cp /usr/mqtt2mysql/flink-con-mqtt.jar com.iniesta.flink.connector.mqtt.Mqtt2Mysql --conf /conf/conf.properties

 


<|start_filename|>src/main/java/com/iniesta/flink/connector/mqtt/MqttWordCount2Mqtt.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class MqttWordCount2Mqtt {

	//
	// Program
	//

	public static void main(String[] args) throws Exception {

		if (args.length != 3) {
			System.err.println("USAGE:\nMqttWordCount2Mqtt <hostname> <topic IN> <topic OUT>");
			return;
		}

		String hostName = args[0];
		String topicIn = args[1];
		String topicOut = args[2];

		// set up the execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// get input data
		DataStream<MqttMessage> text = env.addSource(new MqttSource(hostName, topicIn));

		DataStream<Tuple2<String, Integer>> counts =
				// split up the lines in pairs (2-tuples) containing: (word,1)
				text.flatMap(new LineSplitter())
						// group by the tuple field "0" and sum up tuple field "1"
						.keyBy(0).sum(1);

		// counts.print();
		counts.addSink(new MqttSink<>(hostName, topicOut));

		// execute program
		env.execute("Java WordCount from SocketTextStream Example");
	}

	//
	// User Functions
	//

	/**
	 * Implements the string tokenizer that splits sentences into words as a
	 * user-defined FlatMapFunction. The function takes a line (String) and splits
	 * it into multiple pairs in the form of "(word,1)" (Tuple2<String, Integer>).
	 */
	@SuppressWarnings("serial")
	public static final class LineSplitter implements FlatMapFunction<MqttMessage, Tuple2<String, Integer>> {

		@Override
		public void flatMap(MqttMessage value, Collector<Tuple2<String, Integer>> out) {
			// normalize and split the line
			String[] tokens = value.getPayload().toLowerCase().split("\\W+");

			// emit the pairs
			for (String token : tokens) {
				if (token.length() > 0) {
					out.collect(new Tuple2<String, Integer>(token, 1));
				}
			}
		}
	}
}


<|start_filename|>src/main/java/com/iniesta/flink/connector/mqtt/MqttSink.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;

public class MqttSink<T> extends RichSinkFunction<T> {

	private static final long serialVersionUID = 7883296716671354462L;

	private transient BlockingConnection blockingConnection;
	private String topic;
	private QoS qos;
	private boolean retain;

	private String host;

	private int port;

	public MqttSink(String host, String topic) {
		this(host, 1883, topic, QoS.AT_LEAST_ONCE, false);
	}

	
	public MqttSink(String host, int port, String topic)  {
		this(host, port, topic, QoS.AT_LEAST_ONCE, false);
	}
	
	public MqttSink(String host, int port, String topic, QoS qos, boolean retain) {
		this.host = host;
		this.port = port;		
		this.topic = topic;
		this.qos = qos;
		this.retain = retain;
	}
	
	@Override
	public void invoke(T event) throws Exception {
		byte[] payload = event.toString().getBytes();
		blockingConnection.publish(topic, payload, qos, retain);
	}

	@Override
	public void close() throws Exception {
		super.close();
		blockingConnection.disconnect();
	}

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		MQTT mqtt = new MQTT();
		mqtt.setHost(host, port);
		blockingConnection = mqtt.blockingConnection();
		blockingConnection.connect();
	}

	
	
}


<|start_filename|>src/main/java/com/iniesta/com/SimpleDataQuery.java<|end_filename|>
package com.iniesta.com;

import java.io.FileInputStream;
import java.util.List;
import java.util.Properties;

import com.iniesta.flink.connector.mysql.MysqlDao;

public class SimpleDataQuery {

	public static void main(String[] args) throws Exception {
		Properties props = new Properties();
		props.load(new FileInputStream(args[0]));
		
		MysqlDao dao = new MysqlDao(props);
		dao.open();
		
		List<String> res = dao.testQuery();
		System.out.println(res);
		
		dao.close();
		
	}
}


<|start_filename|>src/main/java/com/iniesta/flink/connector/mqtt/Mqtt2Mysql.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

import java.io.FileInputStream;
import java.util.Properties;

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import com.iniesta.flink.connector.mysql.MysqlSink;


public class Mqtt2Mysql {

	//
	//	Program
	//

	public static void main(String[] args) throws Exception {

		ParameterTool pt = ParameterTool.fromArgs(args);
		
		String properties = pt.getRequired("conf");
		Properties props = new Properties();
		props.load(new FileInputStream(properties));
		
		String hostName = props.getProperty("mqtt.host");
		int port = Integer.parseInt(props.getProperty("mqtt.port", "1883"));		
		String topicIn = props.getProperty("mqtt.topic");
		System.out.println("Properties.... " + props.toString());

		// set up the execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment
				.getExecutionEnvironment();

		// get input data
		DataStream<MqttMessage> text = env.addSource(new MqttSource(hostName, port, topicIn));

//		DataStream<Tuple2<String, Integer>> counts =
		// split up the lines in pairs (2-tuples) containing: (word,1)
//		text.flatMap(new LineSplitter())
		// group by the tuple field "0" and sum up tuple field "1"
//				.keyBy(0)
//				.sum(1);

//		text.print();
//		counts.addSink(new MqttSink<>(hostName, topicOut));
		text.addSink(new MysqlSink(props));
		// execute program		
		int parallelism = Integer.parseInt(props.getProperty("flink.parallelism", "2"));
		env.setParallelism(parallelism);
		System.out.println("About to execute....");
		env.execute("Mqtt 2 JDBC");
	}

	//
	// 	User Functions
	//

	/**
	 * Implements the string tokenizer that splits sentences into words as a user-defined
	 * FlatMapFunction. The function takes a line (String) and splits it into
	 * multiple pairs in the form of "(word,1)" (Tuple2<String, Integer>).
	 */
	@SuppressWarnings("serial")
	public static final class LineSplitter implements FlatMapFunction<MqttMessage, Tuple2<String, Integer>> {

		@Override
		public void flatMap(MqttMessage value, Collector<Tuple2<String, Integer>> out) {
			// normalize and split the line
			String[] tokens = value.getPayload().toLowerCase().split("\\W+");

			// emit the pairs
			for (String token : tokens) {
				if (token.length() > 0) {
					out.collect(new Tuple2<String, Integer>(token, 1));
				}
			}
		}
	}	
}


<|start_filename|>src/test/java/com/iniesta/flink/connector/mqtt/DummyTest.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

import org.junit.Assert;

public class DummyTest {

	public void test1() {
		Assert.assertTrue(true);
	}
}


<|start_filename|>src/test/java/com/iniesta/flink/connector/mqtt/BrokerPublisherManual.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

import java.net.URISyntaxException;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.Message;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;

public class BrokerPublisherManual {
	
	public static void main(String[] args) throws Exception {
		Runnable runnable = new Runnable() {
			
			@Override
			public void run() {
				try {
					publish(1000);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		};
		ExecutorService pool = Executors.newFixedThreadPool(3);
		for (int i = 0; i < 50; i++) {
			pool.submit(runnable);
		}
		
	}

	
	
	private static void publish(int num) throws URISyntaxException, Exception {
		MQTT mqtt = new MQTT();
		mqtt.setHost("cydonia", 1883);
		
		Random random = new Random();
		BlockingConnection blockingConnection = mqtt.blockingConnection();
		
		blockingConnection.connect();
		for (int i = 0; i < num; i++) {
			String payload = "Hello " + random.nextInt(10); 
			blockingConnection.publish("/samples/in", payload.getBytes(), QoS.AT_LEAST_ONCE, false);
		}
		
		blockingConnection.disconnect();
	}
}


<|start_filename|>src/main/java/com/iniesta/flink/connector/mqtt/MqttOutputFormat.java<|end_filename|>
package com.iniesta.flink.connector.mqtt;

import java.io.IOException;
import java.net.URISyntaxException;

import org.apache.flink.api.common.io.RichOutputFormat;
import org.apache.flink.configuration.Configuration;
import org.fusesource.mqtt.client.BlockingConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;

public class MqttOutputFormat<IT> extends RichOutputFormat<IT>{

	private static final long serialVersionUID = -260008582961487797L;

	private transient BlockingConnection blockingConnection;
	private transient MQTT mqtt;
	private String topic;
	private QoS qos;
	private boolean retain;
	private String host;
	private int port;

	public MqttOutputFormat(String host, String topic) {
		this(host, 1883, topic, QoS.AT_LEAST_ONCE, false);
	}
	
	public MqttOutputFormat(String host, int port, String topic) {
		this(host, port, topic, QoS.AT_LEAST_ONCE, false);
	}
	
	public MqttOutputFormat(String host, int port, String topic, QoS qos, boolean retain) {
		this.host = host;
		this.port = port;		
		this.topic = topic;
		this.qos = qos;
		this.retain = retain;
	}
	
	@Override
	public void close() throws IOException {
		System.out.println("closing....");
		try {
			if(blockingConnection!=null) {
				blockingConnection.disconnect();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	@Override
	public void configure(Configuration arg0) {
		System.out.println("configuring output format....");
	}

	@Override
	public void open(int arg0, int arg1) throws IOException {
		System.out.println("opening...."+ arg0 + " " + arg1);
		mqtt = new MQTT();
		try {
			mqtt.setHost(host, port);
			blockingConnection = mqtt.blockingConnection();
			blockingConnection.connect();
		} catch (URISyntaxException e) {
			e.printStackTrace();
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	}

	@Override
	public void writeRecord(IT record) throws IOException {
		System.out.println("writing record...." + record);
		byte[] payload = record.toString().getBytes();
		try {
			blockingConnection.publish(topic, payload, qos, retain);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}
