var open = amqp.connect();
open
  .then(function (conn) {
    process.once("SIGINT", conn.close.bind(conn));
    return doStuffWithConnection(conn);
  })
  .then(null, console.warn);

var dom = domain.create();
dom.on("error", gracefullyRestart);

dom.run(function () {
  amqp.connect(function (err, conn) {
    // ...
  });
});

var open = amqp.connect();
open
  .then(function (conn) {
    var ok = doStuffWithConnection(conn);
    return ok.then(conn.close.bind(conn));
  })
  .then(null, console.warn);

amqp
  .connect()
  .then(
    function (conn) {
      // Everything ok, but ..
      throw new Error("SNAFU");
    },
    function (err) {
      console.error("Connect failed: %s", err);
    }
  )
  .then(null, function (err) {
    console.error(
      "Connect succeeded, but error thrown: %s",
      err
    );
  });

amqp.connect(function (err, conn) {
  if (err !== null) return console.warn(err);
  doStuffWithConnection(conn, function () {
    conn.close();
  });
});

require("amqplib/callback_api").connect(function (err, c) {
  c.createConfirmChannel(function (err, ch) {
    ch.sendToQueue(
      "foo",
      new Buffer("foobar"),
      {},
      function (err, ok) {
        if (err !== null) console.warn("Message nacked!");
        else console.log("Message acked");
      }
    );
  });
});

amqp
  .connect()
  .then(function (conn) {
    var ok = conn.createChannel();
    ok = ok.then(function (ch) {
      return when.all([
        ch.assertQueue("foo"),
        ch.assertExchange("bar"),
        ch.bindQueue("foo", "bar", "baz"),
        ch.consume("foo", handleMessage),
      ]);
    });
    return ok;
  })
  .then(null, console.warn);

var open = require("amqplib").connect();
open.then(function (c) {
  c.createConfirmChannel().then(function (ch) {
    ch.sendToQueue(
      "foo",
      new Buffer("foobar"),
      {},
      function (err, ok) {
        if (err !== null) console.warn("Message nacked!");
        else console.log("Message acked");
      }
    );
  });
});

var dom = domain.create();
dom.on("error", gracefullyRestart);

amqp.connect(function (err, conn) {
  dom.add(conn);
  //...
});

await consumer.connect();

await consumer.subscribe({ topic: "topic-A" });

// Subscribe can be called several times
await consumer.subscribe({ topic: "topic-B" });
await consumer.subscribe({ topic: "topic-C" });

// It's possible to start from the beginning of the topic
await consumer.subscribe({
  topic: "topic-D",
  fromBeginning: true,
});

const { Kafka } = require("kafkajs");
const kafka = new Kafka({
  clientId: "my-app",
  brokers: ["kafka1:9092", "kafka2:9092"],
});
const producer = kafka.producer();
const consumer = kafka.consumer({ groupId: "test-group" });
const run = async () => {
  // Producing
  await producer.connect();
  await producer.send({
    topic: "test-topic",
    messages: [{ value: "Hello KafkaJS user!" }],
  }); // Consuming
  await consumer.connect();
  await consumer.subscribe({
    topic: "test-topic",
    fromBeginning: true,
  });
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      console.log({
        partition,
        offset: message.offset,
        value: message.value.toString(),
      });
    },
  });
};
run().catch(console.error);

await consumer.connect();
await consumer.subscribe({ topic: /topic-(eu|us)-.*/i });

await consumer.connect();
await consumer.subscribe({ topic: "example" });

// you don't need to await consumer#run
consumer.run({
  eachMessage: async ({ topic, message }) => true,
});
consumer.seek({
  topic: "example",
  partition: 0,
  offset: 12384,
});

await consumer.connect();
await consumer.subscribe({ topic: "jobs" });

await consumer.run({
  eachMessage: async ({ topic, message }) => {
    try {
      await sendToDependency(message);
    } catch (e) {
      if (e instanceof TooManyRequestsError) {
        consumer.pause([{ topic }]);
        setTimeout(
          () => consumer.resume([{ topic }]),
          e.retryAfter * 1000
        );
      }

      throw e;
    }
  },
});

const consumer = kafka.consumer({ groupId: "test-group" });

await consumer.connect();
await consumer.subscribe({
  topic: "test-topic",
  fromBeginning: true,
});

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    console.log({
      value: message.value.toString(),
    });
  },
});

const kafka = new Kafka(...)
const admin = kafka.admin()

// remember to connect and disconnect when you are done
await admin.connect()
await admin.disconnect()

const kafka = new Kafka(...)
const admin = kafka.admin()

// remember to connect and disconnect when you are done
await admin.connect()
await admin.disconnect()