with Ably and Kafka?
Ably API provides a platform for adding real-time messaging, streaming, and presence capabilities to your apps. With Pipedream, you can harness this power to create dynamic, event-driven workflows. Automate notifications, synchronize data across systems in real time, or react to Ably events to trigger actions in other services, like sending emails, updating databases, or calling external APIs.
import { axios } from "@pipedream/platform"
export default defineComponent({
props: {
ably: {
type: "app",
app: "ably",
}
},
async run({steps, $}) {
return await axios($, {
url: `https://rest.ably.io/channels/rest-example/messages`,
auth: {
username: `${this.ably.$auth.api_key_initial}`,
password: `${this.ably.$auth.api_key_remaining}`,
},
})
},
})
import { Kafka } from "kafkajs"
export default defineComponent({
props: {
kafka: {
type: "app",
app: "kafka",
}
},
async run({steps, $}) {
const kafka = new Kafka({
brokers: [`${this.kafka.$auth.host}:${this.kafka.$auth.port}`],
});
const consumer = kafka.consumer({ groupId: 'TestGroup' });
await consumer.connect()
await consumer.subscribe({ topic: 'SampleTopic', fromBeginning: true });
let consumedMessage = "";
const eachMessage = async function({ topic, partition, message }){
consumedMessage = message.value.toString();
return consumedMessage;
};
await consumer.run({
eachMessage,
});
const producer = kafka.producer();
await producer.connect()
await producer.send({
topic: 'SampleTopic',
messages: [
{ value: 'Welcome KafkaJS + Pipedream users! '+ new Date().toISOString() },
],
});
await producer.disconnect();
const data = await consumer.describeGroup();
return { consumedMessage, groupDescription: data };
},
})