Adding Tools to the DAP and DTS, Overview and Examples

Introduction

Prerequisites

Polyglot

Medici

• Java

  Connecting to RabbitMQ

  protected void startExtractor(String rabbitMQUsername,
  	String rabbitMQpassword) {
  	try{ 
   		//Open channel and declare exchange and consumer
  		ConnectionFactory factory = new ConnectionFactory();
  		factory.setHost(serverAddr);
  		factory.setUsername(rabbitMQUsername);
  		factory.setPassword(rabbitMQpassword);
  		Connection connection = factory.newConnection();
  
   		final Channel channel = connection.createChannel();
  		channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
  
  		channel.queueDeclare(QUEUE_NAME,DURABLE,EXCLUSIVE,AUTO_DELETE,null);
  		channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.file.text.plain.#");
   
   		this.channel = channel;
  
   		// create listener
  		channel.basicConsume(QUEUE_NAME, false, CONSUMER_TAG, new DefaultConsumer(channel) {
   			@Override
   			public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  				messageReceived = new String(body);
   				long deliveryTag = envelope.getDeliveryTag();
   				// (process the message components here ...)
  				System.out.println(" {x} Received '" + messageReceived + "'");
   
  				replyProps = new AMQP.BasicProperties.Builder().correlationId(properties.getCorrelationId()).build();
  				replyTo = properties.getReplyTo();
   
  				processMessageReceived();
  				System.out.println(" [x] Done");
  				channel.basicAck(deliveryTag, false);
  			}
  		});
  
   		// start listening 
  		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
   		while (true) {
  			Thread.sleep(1000);
  		}
  	}
   	catch(Exception e){
  		e.printStackTrace();
  		System.exit(1);
  	} 
  }

   

• C++

• Python

   

  Instantiating the logger and starting the extractor

  def main():
   global logger
  
   # name of receiver
  receiver='ExamplePythonExtractor'
  
   # configure the logging system
  logging.basicConfig(format="%(asctime)-15s %(name)-10s %(levelname)-7s : %(message)s", level=logging.WARN)
  logger = logging.getLogger(receiver)
  logger.setLevel(logging.DEBUG)
   
   if len(sys.argv) != 4:
  logger.info("Input RabbitMQ username, followed by RabbitMQ password and Medici REST API key.")
  sys.exit()
   
   global playserverKey
  playserverKey = sys.argv[3]

  Connecting to RabbitMQ

  # connect to rabbitmq using input username and password 
  credentials = pika.PlainCredentials(sys.argv[1], sys.argv[2])
  parameters = pika.ConnectionParameters(credentials=credentials)
  connection = pika.BlockingConnection(parameters)
   
   # connect to channel
  channel = connection.channel()
  
   # declare the exchange
  channel.exchange_declare(exchange='medici', exchange_type='topic', durable=True)
  
   # declare the queue
  channel.queue_declare(queue=receiver, durable=True)
  
   # connect queue and exchange
  channel.queue_bind(queue=receiver, exchange='medici', routing_key='*.file.text.plain')
  
   # create listener
  channel.basic_consume(on_message, queue=receiver, no_ack=False)
  
   # start listening
  logger.info("Waiting for messages. To exit press CTRL+C")
   try:
  channel.start_consuming()
   except KeyboardInterrupt:
  channel.stop_consuming()
  

Versus