with Ambient Weather and Kafka?
The Ambient Weather API provides real-time access to personal weather station data, enabling you to monitor and analyze environmental conditions. Through Pipedream's integration, you can automate actions based on this data, such as triggering alerts when specific weather thresholds are reached, logging climate trends for analysis, or controlling smart home devices to adapt to changing conditions.
import { axios } from "@pipedream/platform"
export default defineComponent({
props: {
ambient_weather: {
type: "app",
app: "ambient_weather",
}
},
async run({steps, $}) {
return await axios($, {
url: `https://${this.ambient_weather.$auth.subdomain}.ambientweather.net/v1/devices`,
params: {
applicationKey: `${this.ambient_weather.$auth.application_key}`,
apiKey: `${this.ambient_weather.$auth.api_key}`,
},
})
},
})
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 };
},
})