Integrating with Client Libraries

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

channel.queueDeclare("hello", false, false, false, null);
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

channel.close();
connection.close();

String message = "Hello World!";
channel.basicPublish("", "hello", null, message.getBytes("UTF-8"));
System.out.println(" [x] Sent '" + message + "'");

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    String message = new String(delivery.getBody(), "UTF-8");
    System.out.println(" [x] Received '" + message + "'");
};
CancelCallback cancelCallback = consumerTag -> {
    System.out.println("Consumer was cancelled");
};

channel.basicConsume("hello", true, deliverCallback, cancelCallback);

import pika

# Establish connection
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare queue
channel.queue_declare(queue='hello')

# Publish message
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='Hello World!')

print(" [x] Sent 'Hello World!'")
connection.close()

import pika

# Establish connection
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare queue
channel.queue_declare(queue='hello')

# Define callback function
def callback(ch, method, properties, body):
    print(f" [x] Received {body}")

# Set up consumer
channel.basic_consume(queue='hello',
                      on_message_callback=callback,
                      auto_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

const amqp = require("amqplib");

async function connect() {
  const connection = await amqp.connect("amqp://localhost");
  const channel = await connection.createChannel();
  await channel.assertQueue("my-queue", { durable: true });
  console.log("Connected and queue declared");
}

connect().catch(console.error);

const amqp = require("amqplib");

async function publishAndConsume() {
  const connection = await amqp.connect("amqp://localhost");
  const channel = await connection.createChannel();

  const queue = "task-queue";
  await channel.assertQueue(queue, { durable: true });

  // Publish a message
  const msg = "Hello RabbitMQ!";
  channel.sendToQueue(queue, Buffer.from(msg), { persistent: true });
  console.log(" [x] Sent " + msg);

  // Consume messages
  channel.consume(queue, (msg) => {
    console.log(" [x] Received " + msg.content.toString());
    channel.ack(msg);
  }, { noAck: false });
}

publishAndConsume().catch(console.error);

package main

import (
    "fmt"
    "log"

    "github.com/rabbitmq/amqp091-go"
)

func main() {
    conn, err := amqp091.Dial("amqp://guest:guest@localhost:5672/")
    if err != nil {
        log.Fatalf("Failed to connect to RabbitMQ: %v", err)
    }
    defer conn.Close()

    ch, err := conn.Channel()
    if err != nil {
        log.Fatalf("Failed to open a channel: %v", err)
    }
    defer ch.Close()

    fmt.Println("Connected to RabbitMQ successfully")
}

@Configuration
public class RabbitConfig {

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }
}

@Service
public class MessageSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend("my.exchange", "my.routing.key", message);
    }
}

@Component
public class MessageReceiver {

    @RabbitListener(queues = "my.queue")
    public void receiveMessage(String message) {
        System.out.println("Received: " + message);
    }
}

import com.rabbitmq.client.ConnectionFactory;

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setConnectionTimeout(30000); // 30 seconds
factory.setRequestedHeartbeat(30);   // 30 seconds
factory.setHandshakeTimeout(15000);  // 15 seconds

Connection connection = factory.newConnection();

const amqp = require('amqplib');

async function connectWithReconnect() {
  const connectionUrl = 'amqp://localhost';
  let conn;

  while (true) {
    try {
      conn = await amqp.connect(connectionUrl);
      console.log('Connected to RabbitMQ');
      break;
    } catch (err) {
      console.error('Connection failed, retrying...', err);
      await new Promise(resolve => setTimeout(resolve, 5000));
    }
  }

  return conn;
}

connectWithReconnect().then(conn => {
  // Use connection...
});

const amqp = require('amqplib');

async function handleClientErrors() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();

    // Simulate a client error
    await channel.assertQueue('nonexistent_queue');
  } catch (error) {
    console.error('Client error occurred:', error.message);
    // Handle specific error types
    if (error.code === 'ENOTFOUND') {
      console.error('Connection failed - check RabbitMQ server');
    } else if (error.code === 'ECONNREFUSED') {
      console.error('Connection refused - server not running');
    }
  }
}

const amqp = require('amqplib');

async function connectWithMetrics() {
  const connection = await amqp.connect('amqp://localhost', {
    clientProperties: {
      'client-id': 'my-app',
      'metrics-enabled': true,
      'version': '1.0.0'
    }
  });

  const channel = await connection.createChannel();
  return { connection, channel };
}

connectWithMetrics().then(({ connection, channel }) => {
  console.log('Connected with metrics enabled');
});

const amqp = require('amqplib');

async function testIntegration() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    
    const queue = 'test-queue';
    await channel.assertQueue(queue, { durable: false });
    
    const message = 'Hello RabbitMQ';
    channel.sendToQueue(queue, Buffer.from(message));
    
    console.log('Message sent:', message);
    await channel.close();
    await connection.close();
  } catch (error) {
    console.error('Integration test failed:', error);
  }
}

testIntegration();

const amqp = require('amqplib');

async function createConnectionFactory() {
  const connection = await amqp.connect('amqp://localhost');
  const channel = await connection.createChannel();
  
  return {
    connection,
    channel
  };
}

module.exports = { createConnectionFactory };