package com.github.pukkaone.gelf.protocol;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.net.*;
import java.nio.ByteBuffer;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.Date;
import java.util.UUID;

public class GelfAMQPSender extends GelfSender {

    private boolean shutdown = false;

    private ConnectionFactory factory;
    private Connection connection;
    private Channel channel;

    private String exchangeName;
    private String routingKey;
    private int maxRetries;

    public GelfAMQPSender(
            String host, String exchangeName, String routingKey, int maxRetries)
        throws IOException, URISyntaxException, NoSuchAlgorithmException, KeyManagementException
    {
        factory = new ConnectionFactory();
        factory.setUri(host);

        this.exchangeName = exchangeName;
        this.routingKey = routingKey;
        this.maxRetries = maxRetries;
    }

    private ByteBuffer toAMQPBuffer(String json) {
        byte[] messageBytes = gzipMessage(json);
        ByteBuffer buffer = ByteBuffer.allocate(messageBytes.length);
        buffer.put(messageBytes);
        buffer.flip();
        return buffer;
    }

    public boolean sendMessage(GelfMessage message) {
        if (shutdown || !message.isValid()) {
            return false;
        }

        // set unique id to identify duplicates after connection failure
        String messageId = "gelf" + message.getTimestampMillis() + UUID.randomUUID();

        int tries = 0;
        do {
            try {
                // establish the connection the first time
                if (channel == null) {
                    connection = factory.newConnection();
                    channel = connection.createChannel();
                }

                BasicProperties.Builder propertiesBuilder = new BasicProperties.Builder();
                propertiesBuilder.contentType("application/json; charset=utf-8");
                propertiesBuilder.contentEncoding("gzip");
                propertiesBuilder.messageId(messageId);
                propertiesBuilder.timestamp(new Date(message.getTimestampMillis()));
                BasicProperties properties = propertiesBuilder.build();

                channel.basicPublish(
                    exchangeName,
                    routingKey,
                    properties,
                    toAMQPBuffer(message.toJson()).array());
                channel.waitForConfirms();

                return true;
            } catch (Exception e) {
                channel = null;
                tries++;
            }
        } while (tries <= maxRetries || maxRetries < 0);

        return false;
    }

    public void close() {
        shutdown = true;
        try {
            channel.close();
        } catch (Exception e) {
            // Ignore exception closing channel.
        }

        try {
            connection.close();
        } catch (Exception e) {
            // Ignore exception closing connection.
        }
    }
}

@@ -483,16 +483,25 @@
 private 
+volatile 
 boolean 
@@ -518,16 +518,16 @@
  false;%0A
-
 %0A    pri
@@ -1122,24 +1122,1029 @@
 ies;%0A    %7D%0A%0A
+    private synchronized Connection getConnection() throws IOException %7B%0A        if (connection == null) %7B%0A            connection = factory.newConnection();%0A        %7D%0A        return connection;%0A    %7D%0A%0A    private synchronized void closeConnection() %7B%0A        if (connection != null) %7B%0A            try %7B%0A                connection.close();%0A            %7D catch (IOException e) %7B%0A                // Ignore exception closing connection.%0A            %7D%0A            connection = null;%0A        %7D%0A    %7D%0A%0A    private synchronized Channel getChannel() throws IOException %7B%0A        if (channel == null) %7B%0A            channel = getConnection().createChannel();%0A            channel.confirmSelect();%0A        %7D%0A        return channel;%0A    %7D%0A%0A    private synchronized void closeChannel() %7B%0A        if (channel != null) %7B%0A            try %7B%0A                channel.close();%0A            %7D catch (IOException e) %7B%0A                // Ignore exception closing channel.%0A            %7D%0A            channel = null;%0A        %7D%0A    %7D%0A%0A
     private 
@@ -2759,223 +2759,39 @@
     
-// establish the connection the first time%0A                if (channel == null) %7B%0A                    connection = factory.newConnection();%0A                    channel = connection.createChannel();%0A                %7D
+Channel channel = getChannel();
 %0A%0A  
@@ -3517,32 +3517,57 @@
        %7D catch (
+InterruptedException %7C IO
 Exception e) %7B%0A 
@@ -3578,37 +3578,72 @@
            c
+loseC
 hannel
- = null
+();%0A                closeConnection()
 ;%0A          
@@ -3821,274 +3821,50 @@
     
-try %7B%0A            channel.close();%0A        %7D catch (Exception e) %7B%0A            // Ignore exception closing channel.%0A        %7D%0A%0A        try %7B%0A            connection.close();%0A        %7D catch (Exception e) %7B%0A            // Ignore exception closing connection.%0A        %7D
+closeChannel();%0A        closeConnection();
 %0A   

Configure AMQP channel for confirms