Envío de eventos con RabbitMQ

$docker run -d --hostname my-rabbit \
    --name some-rabbit \
    -v $(pwd)/rabbit://var/lib/rabbitmq/mnesia/rabbit \
    -p 5672:5672 -p 15672:15672 \
    rabbitmq:3-management

exchangeName="groovy-script"
queueName="grabbit"
routingKey='#'

factory = new ConnectionFactory()
	factory.username='guest'
	factory.password='guest'
	factory.virtualHost='/'
	factory.host= args[0] ?: 'localhost'
	factory.port=5672
conn = factory.newConnection()

channel = conn.createChannel()
channel.exchangeDeclare(exchangeName, "direct", true)
channel.queueDeclare(queueName, true, false, false, null)
channel.queueBind(queueName, exchangeName, routingKey)

	monitor = args[2]

	if (new File(monitor).exists() == false)
		new File(monitor).newPrintWriter().println "${new Date()}"

	FileSystemManager manager = VFS.getManager();
	FileObject fobj = manager.resolveFile(new File(monitor).absolutePath)
	DefaultFileMonitor fm = new DefaultFileMonitor(this as FileListener)
	fm.delay = 500
	fm.addFile(fobj)
	fm.start()

void fileChanged(FileChangeEvent evt) throws Exception {

	RandomAccessFile randomAccessFile = new RandomAccessFile(evt.file.localFile, "r");	//(1)
	long fileLength = evt.file.localFile.length() - 1;
	randomAccessFile.seek(fileLength);
	StringBuilder stringBuilder = new StringBuilder()

	for(long pointer = fileLength; pointer >= 0; pointer--){
		randomAccessFile.seek(pointer)
		char c = (char)randomAccessFile.read()
		if(c == '\n' && fileLength == pointer){
			continue
		}
		if(c == '\n' && fileLength != pointer){
			break
		}
		stringBuilder.append(c)
	}
	println "Soy el producer enviando ${stringBuilder.toString().reverse()}"
	channel.basicPublish(exchangeName, routingKey, null, stringBuilder.toString().reverse().bytes) //(2)
}

	boolean autoAck = true;
	channel.basicConsume(queueName, autoAck, new DefaultConsumer(channel) {
         public void handleDelivery(String consumerTag,
                                    Envelope envelope,
                                    AMQP.BasicProperties properties,
                                    byte[] body)
             throws IOException{
		println "Soy el consumer recibiendo ${new String(body)}"
         }
	});

version: '2'

services:

  #(1)
  rabbitmq-container:
    image: rabbitmq:3-management
    hostname: my-rabbit
    volumes:
      - ./rabbit:/var/lib/rabbitmq/mnesia/rabbit
    ports:
      - 5672:5672
      - 15672:15672

  #(2)
  consumer:
    image: groovy:2.4-jdk8
    volumes:
      - ./:/home/groovy/
    command: groovy /home/groovy/grabbit.groovy rabbitmq-container consumer #(3)

  #(4)
  producer1:
      image: groovy:2.4-jdk8
      volumes:
        - ./:/home/groovy/
      command: groovy /home/groovy/grabbit.groovy rabbitmq-container producer /home/groovy/grabbit.log #(5)

  #(4)
  producer2:
      image: groovy:2.4-jdk8
      volumes:
        - ./:/home/groovy/
      command: groovy /home/groovy/grabbit.groovy rabbitmq-container producer /home/groovy/grabbit2.log #(6)

$ docker-compose up

@Grapes([
    @Grab(group='com.rabbitmq', module='amqp-client', version='3.1.2'),
	@Grab(group='org.apache.commons', module='commons-vfs2', version='2.2')
])

import com.rabbitmq.client.*
import groovy.json.*

import org.apache.commons.vfs2.FileChangeEvent
import org.apache.commons.vfs2.FileListener
import org.apache.commons.vfs2.FileObject
import org.apache.commons.vfs2.FileSystemManager
import org.apache.commons.vfs2.VFS
import org.apache.commons.vfs2.impl.DefaultFileMonitor

//tag::conexion[]
exchangeName="groovy-script"
queueName="grabbit"
routingKey='#'

factory = new ConnectionFactory()
	factory.username='guest'
	factory.password='guest'
	factory.virtualHost='/'
	factory.host= args[0] ?: 'localhost'
	factory.port=5672
conn = factory.newConnection()

channel = conn.createChannel()
channel.exchangeDeclare(exchangeName, "direct", true)
channel.queueDeclare(queueName, true, false, false, null)
channel.queueBind(queueName, exchangeName, routingKey)
//end::conexion[]

if( args[1] == 'producer' ) {

	//tag::producer[]
	monitor = args[2]

	if (new File(monitor).exists() == false)
		new File(monitor).newPrintWriter().println "${new Date()}"

	FileSystemManager manager = VFS.getManager();
	FileObject fobj = manager.resolveFile(new File(monitor).absolutePath)
	DefaultFileMonitor fm = new DefaultFileMonitor(this as FileListener)
	fm.delay = 500
	fm.addFile(fobj)
	fm.start()
	//end::producer[]

}

if( args[1] == 'consumer' ) {
	//tag::consumer[]
	boolean autoAck = true;
	channel.basicConsume(queueName, autoAck, new DefaultConsumer(channel) {
         public void handleDelivery(String consumerTag,
                                    Envelope envelope,
                                    AMQP.BasicProperties properties,
                                    byte[] body)
             throws IOException{
		println "Soy el consumer recibiendo ${new String(body)}"
         }
	});
	//end::consumer[]
}

//tag::filechanged[]
void fileChanged(FileChangeEvent evt) throws Exception {

	RandomAccessFile randomAccessFile = new RandomAccessFile(evt.file.localFile, "r");	//(1)
	long fileLength = evt.file.localFile.length() - 1;
	randomAccessFile.seek(fileLength);
	StringBuilder stringBuilder = new StringBuilder()

	for(long pointer = fileLength; pointer >= 0; pointer--){
		randomAccessFile.seek(pointer)
		char c = (char)randomAccessFile.read()
		if(c == '\n' && fileLength == pointer){
			continue
		}
		if(c == '\n' && fileLength != pointer){
			break
		}
		stringBuilder.append(c)
	}
	println "Soy el producer enviando ${stringBuilder.toString().reverse()}"
	channel.basicPublish(exchangeName, routingKey, null, stringBuilder.toString().reverse().bytes) //(2)
}
//end::filechanged[]

void fileCreated(FileChangeEvent arg0) throws Exception {

}

void fileDeleted(FileChangeEvent arg0) throws Exception {

}