Get started with the Starlight for RabbitMQ extension

helm repo add datastax-pulsar https://datastax.github.io/pulsar-helm-chart

VALUES_URL="https://raw.githubusercontent.com/datastaxdevs/luna-streaming-examples/main/starlight-for-rabbitmq/values.yaml"
helm install \
  --namespace datastax-pulsar \
  --create-namespace \
  --values $VALUES_URL \
  --version 3.0.4 \
  my-pulsar-cluster \
  datastax-pulsar/pulsar

kubectl -n datastax-pulsar wait --for=condition=Ready pod/pulsar-broker-0 --timeout=120s

tar xvfz starlight-rabbitmq-VERSION-all.tar.gz

messagingProtocols=rabbitmq
protocolHandlerDirectory=./protocols

amqpListeners=amqp://127.0.0.1:5672
advertisedAddress=127.0.0.1

tar xvfz starlight-rabbitmq-VERSION-all.tar.gz

proxyExtensions=rabbitmq
proxyExtensionsDirectory=./proxyextensions

amqpListeners=amqp://127.0.0.1:5672
advertisedAddress=127.0.0.1

tar xvfz starlight-rabbitmq-VERSION-all.tar.gz

brokerServiceURL=pulsar://localhost:6650
brokerWebServiceURL=http://localhost:8080
configurationStoreServers=localhost:2181

java -jar ./starlight-rabbitmq/target/starlight-rabbitmq-${version}-jar-with-dependencies.jar -c conf/starlight-for-rabbitmq.conf

tar xvfz starlight-rabbitmq-VERSION-all.tar.gz

messagingProtocols=rabbitmq
protocolHandlerDirectory=./protocols

amqpListeners=amqp://127.0.0.1:5672
advertisedAddress=127.0.0.1

tar xvfz starlight-rabbitmq-VERSION-all.tar.gz

proxyExtensions=rabbitmq
proxyExtensionsDirectory=./proxyextensions

amqpListeners=amqp://127.0.0.1:5672
advertisedAddress=127.0.0.1

tar xvfz starlight-rabbitmq-VERSION-all.tar.gz

brokerServiceURL=pulsar://localhost:6650
brokerWebServiceURL=http://localhost:8080
configurationStoreServers=localhost:2181

java -jar ./starlight-rabbitmq/target/starlight-rabbitmq-${version}-jar-with-dependencies.jar -c conf/starlight-for-rabbitmq.conf

import ssl
import pika

virtual_host = "VIRTUAL_HOST"
token = "PASSWORD"

context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
context.verify_mode = ssl.CERT_NONE
context.check_hostname = False
context.load_default_certs()
ssl_options = pika.SSLOptions(context)

connection = pika.BlockingConnection(pika.ConnectionParameters(
    virtual_host=virtual_host,
    host="HOST",
    ssl_options=ssl_options,
    port=PORT,
    credentials=pika.PlainCredentials("", token)))
print("connection success")

channel = connection.channel()
print("started a channel")

channel.queue_declare(queue='queuename')

for x in range(10):
    channel.basic_publish(exchange='',
                      routing_key='routingkey',
                      body='message body goes here')
    print(" sent one")

connection.close()

python3 connect-test.py

connection success
started a channel
 sent one
 sent one
 sent one
 sent one
 sent one
 sent one
 sent one
 sent one
 sent one
 sent one

mvn archetype:generate \
    -DgroupId=org.example \
    -DartifactId=StarlightForRabbitMqClient \
    -DarchetypeArtifactId=maven-archetype-quickstart \
    -DinteractiveMode=false

cd StarlightForRabbitMqClient

<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.16.0</version>
</dependency>

package org.example;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeoutException;

public class App {
  private static final String username = "";
  private static final String password = "PASSWORD";
  private static final String host = "HOST";
  private static final int port = 5671;
  private static final String virtual_host = "VIRTUAL_HOST";
  private static final String queueName = "queuename";
  private static final String amqp_URI = String.format("amqps://%s:%s@%s:%d/%s", username, password, host, port, virtual_host.replace("/","%2f"));

  public static void main(String[] args) throws IOException, TimeoutException, URISyntaxException, NoSuchAlgorithmException, KeyManagementException, InterruptedException {

    ConnectionFactory factory = new ConnectionFactory();
    factory.setUri(amqp_URI);

    /*
    You could also set each value individually
      factory.setHost(host);
      factory.setPort(port);
      factory.setUsername(username);
      factory.setPassword(password);
      factory.setVirtualHost(virtual_host);
      factory.useSslProtocol();
     */

    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.queueDeclare(queueName, false, false, false, null);

    String publishMessage = "Hello World!";
    channel.basicPublish("", queueName, null, publishMessage.getBytes());
    System.out.println(" Sent '" + publishMessage + "'");

    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
      String consumeMessage = new String(delivery.getBody(), StandardCharsets.UTF_8);
      System.out.println(" Received '" + consumeMessage + "'");
    };

    channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });

    Thread.sleep(4000); // wait a bit for messages to be received

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

mvn clean package assembly:single
java -jar target/StarlightForRabbitMqClient-1.0-SNAPSHOT-jar-with-dependencies.jar

Sent 'Hello World!'
Received 'Hello World!'

#!/usr/bin/env python
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(port=5672))
channel = connection.channel()

try:
    channel.queue_declare("test-queue")
    print("created test-queue queue")

    channel.basic_publish(exchange="", routing_key="test-queue", body="test".encode('utf-8'))
    print("published message test")

    _, _, res = channel.basic_get(queue="test-queue", auto_ack=True)
    assert res is not None, "should have received a message"
    print("received message: " + res.decode())

    channel.queue_delete("test-queue")
    print("deleted test-queue queue")

finally:
    connection.close()

python ./test-queue.py

created test-queue queue
published message test
received message: test
deleted test-queue queue

mvn archetype:generate \
    -DgroupId=org.example \
    -DartifactId=StarlightForRabbitMqClient \
    -DarchetypeArtifactId=maven-archetype-quickstart \
    -DinteractiveMode=false

cd StarlightForRabbitMqClient

<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>amqp-client</artifactId>
  <version>5.16.0</version>
</dependency>

package org.example;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.concurrent.TimeoutException;

public class App {
  private static final String queueName = "queuename";

  public static void main(String[] args) throws IOException, TimeoutException, URISyntaxException, NoSuchAlgorithmException, KeyManagementException, InterruptedException {

    // Use the default values to connect on localhost:5672
    ConnectionFactory factory = new ConnectionFactory();
    factory.setUsername(userName);
    factory.setPassword(password);
    factory.setVirtualHost(virtualHost);
    factory.setHost(hostName);
    factory.setPort(portNumber);

    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.queueDeclare(queueName, false, false, false, null);

    String publishMessage = "Hello World!";
    channel.basicPublish("", queueName, null, publishMessage.getBytes());
    System.out.println(" Sent '" + publishMessage + "'");

    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
      String consumeMessage = new String(delivery.getBody(), StandardCharsets.UTF_8);
      System.out.println(" Received '" + consumeMessage + "'");
    };

    channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });

    Thread.sleep(4000); // wait a bit for messages to be received

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

mvn clean package assembly:single
java -jar target/StarlightForRabbitMqClient-1.0-SNAPSHOT-jar-with-dependencies.jar

Sent 'Hello World!'
Received 'Hello World!'