Code Samples

C# — Two vhosts / Same queue name

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System.Text;

var fixturesFactory = new ConnectionFactory {
    HostName = "your-rabbitmq-host",
    Port = 5671,
    UserName = "fixtures-username",
    Password = "fixtures-password",
    VirtualHost = "fixtures",
    Ssl = new SslOption { Enabled = true, ServerName = "your-rabbitmq-host" }
};

using var fxConn = fixturesFactory.CreateConnection();
using var fxCh = fxConn.CreateModel();
string qname = $"_{customerPackageId}_";
fxCh.QueueDeclare(qname, durable: true, exclusive: false, autoDelete: false);

var tipsFactory = new ConnectionFactory {
    HostName = "your-rabbitmq-host",
    Port = 5671,
    UserName = "tips-username",
    Password = "tips-password",
    VirtualHost = "tips",
    Ssl = new SslOption { Enabled = true, ServerName = "your-rabbitmq-host" }
};

using var tpConn = tipsFactory.CreateConnection();
using var tpCh = tpConn.CreateModel();
tpCh.QueueDeclare(qname, durable: true, exclusive: false, autoDelete: false);

void Consume(IModel ch, string queue, Action<string> handle) {
    var consumer = new EventingBasicConsumer(ch);
    consumer.Received += (_, ea) => handle(Encoding.UTF8.GetString(ea.Body.ToArray()));
    ch.BasicConsume(queue: queue, autoAck: true, consumer: consumer);
}

Consume(fxCh, qname, json => {
    // Upsert fixture by body.fixtureId (delta-only)
});

Consume(tpCh, qname, json => {
    // Parse header.tipEventType: 1=Added, 2=Removed
    // Join by body.fixtureId; for Added: body.tips[] (Tip objects); for Removed: body.tips[] (IDs)
});

Python (pika) — Two vhosts / Same queue name

import ssl, pika, json, threading

def mk_conn(vhost, user, pwd):
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return pika.BlockingConnection(pika.ConnectionParameters(
        host='your-rabbitmq-host', port=5671, virtual_host=vhost,
        credentials=pika.PlainCredentials(user, pwd),
        ssl_options=pika.SSLOptions(ctx, server_hostname='your-rabbitmq-host')
    ))

def run_consumer(vhost, user, pwd, qname, on_message):
    conn = mk_conn(vhost, user, pwd)
    ch = conn.channel()
    ch.queue_declare(queue=qname, durable=True, exclusive=False, auto_delete=False)
    ch.basic_consume(queue=qname, on_message_callback=on_message, auto_ack=True)
    try:
        ch.start_consuming()
    finally:
        try: conn.close()
        except: pass

def on_fixture(ch, method, props, body):
    m = json.loads(body.decode('utf-8'))
    # upsert by m['body']['fixtureId']

def on_tip(ch, method, props, body):
    m = json.loads(body.decode('utf-8'))
    # kind = m['header'].get('tipEventType') # 1=Added, 2=Removed
    # if kind == 1: iterate Tip objects in m['body']['tips']
    # if kind == 2: remove IDs in m['body']['tips']

qname = f"_{customer_package_id}_"
t_fx = threading.Thread(target=run_consumer, args=('fixtures', 'fixtures-username', 'fixtures-password', qname, on_fixture), daemon=True)
t_tp = threading.Thread(target=run_consumer, args=('tips', 'tips-username', 'tips-password', qname, on_tip), daemon=True)
t_fx.start(); t_tp.start(); t_fx.join(); t_tp.join()

Java — Two vhosts / Same queue name

import com.rabbitmq.client.*;

String q = "_" + customerPackageId + "_";

// /fixtures
ConnectionFactory f1 = new ConnectionFactory();
f1.setHost("your-rabbitmq-host");
f1.setPort(5671);
f1.setUsername("fixtures-username");
f1.setPassword("fixtures-password");
f1.setVirtualHost("fixtures");
f1.useSslProtocol();

Connection fxConn = f1.newConnection();
Channel fxCh = fxConn.createChannel();
fxCh.queueDeclare(q, true, false, false, null);

// /tips
ConnectionFactory f2 = new ConnectionFactory();
f2.setHost("your-rabbitmq-host");
f2.setPort(5671);
f2.setUsername("tips-username");
f2.setPassword("tips-password");
f2.setVirtualHost("tips");
f2.useSslProtocol();

Connection tpConn = f2.newConnection();
Channel tpCh = tpConn.createChannel();
tpCh.queueDeclare(q, true, false, false, null);

DeliverCallback fxCb = (tag, d) -> {
    String json = new String(d.getBody(), "UTF-8");
    // upsert by body.fixtureId (delta-only)
};

DeliverCallback tpCb = (tag, d) -> {
    String json = new String(d.getBody(), "UTF-8");
    // parse header.tipEventType (1=Added, 2=Removed)
    // join by body.fixtureId
};

fxCh.basicConsume(q, true, fxCb, tag -> {});
tpCh.basicConsume(q, true, tpCb, tag -> {});