Hono•4d ago

Post method not working with Queues on CF Worker

Hono version: 4.6.20 Wrangler version: 3.107.2 I've tried a lot but fetch function inside queue is not executing in the worker both deployed version and local dev server.

import { Hono, Env } from 'hono'
import { env } from "hono/adapter";
import { send_to_queue } from '../utils/send-to-queue';

type Bindings = {
  REFERRAL_ENGINE_URL: string;
  DB_QUEUE: Queue;
}

interface QueueDataType {
  data: any;
  ref_url: string;
}

const app = new Hono<{ Bindings: Bindings }>()

app.get('/', (c) => {
  return c.json({ message: "Go Away" }, 200);
})

app.post('/', async (c) => {
  const data = await c.req.json()
  const EP_URL = env(c).EP_URL;
  console.log("Data Received")

  const Q_DATA : QueueDataType = {
    data: data,
    ref_url: EP_URL
  }

  await send_to_queue(Q_DATA, c)

  return c.json({ message: 'Request Received!' })
})

async function queueHandler(batch: MessageBatch<QueueDataType>, c: Context) {
  for (let message of batch.messages) {
    console.log("Data Received from Queue");
    try {
      console.log("Sending data to referral engine");
      console.log(message.body.data);
      console.log(typeof(message.body.data));
      const response = await fetch(`${message.body.ref_url}/api/v1/referral`, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': 'Bearer 9e985a31cd0581672bf8454123662de7'
        },
        body: JSON.stringify(message.body.data),
      });

      if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
      }
      // await send_to_ref(message.body.data, message.body.ref_url);
      message.ack();
    } catch (error) {
      console.error("Error forwarding data from queue:", error);
      console.log("Retrying after 100 seconds");
      message.retry({ delaySeconds: 100 });
    }
  }
}

export default {
  fetch: app.fetch,
  async queue(batch: MessageBatch<QueueDataType>, c: Env){
    queueHandler(batch, c)
  },
};

import { Hono, Env } from 'hono'
import { env } from "hono/adapter";
import { send_to_queue } from '../utils/send-to-queue';

type Bindings = {
  REFERRAL_ENGINE_URL: string;
  DB_QUEUE: Queue;
}

interface QueueDataType {
  data: any;
  ref_url: string;
}

const app = new Hono<{ Bindings: Bindings }>()

app.get('/', (c) => {
  return c.json({ message: "Go Away" }, 200);
})

app.post('/', async (c) => {
  const data = await c.req.json()
  const EP_URL = env(c).EP_URL;
  console.log("Data Received")

  const Q_DATA : QueueDataType = {
    data: data,
    ref_url: EP_URL
  }

  await send_to_queue(Q_DATA, c)

  return c.json({ message: 'Request Received!' })
})

async function queueHandler(batch: MessageBatch<QueueDataType>, c: Context) {
  for (let message of batch.messages) {
    console.log("Data Received from Queue");
    try {
      console.log("Sending data to referral engine");
      console.log(message.body.data);
      console.log(typeof(message.body.data));
      const response = await fetch(`${message.body.ref_url}/api/v1/referral`, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': 'Bearer 9e985a31cd0581672bf8454123662de7'
        },
        body: JSON.stringify(message.body.data),
      });

      if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
      }
      // await send_to_ref(message.body.data, message.body.ref_url);
      message.ack();
    } catch (error) {
      console.error("Error forwarding data from queue:", error);
      console.log("Retrying after 100 seconds");
      message.retry({ delaySeconds: 100 });
    }
  }
}

export default {
  fetch: app.fetch,
  async queue(batch: MessageBatch<QueueDataType>, c: Env){
    queueHandler(batch, c)
  },
};

17 Replies

ambergristle•4d ago

hey! can you add syntax highlighting please? also, can you share the relevant bits of your wrangler?

TheLegendVibesOP•4d ago

Here's my wrangler.json

{
  "name": "my-worker",
  "workers_dev": true,
  "vars": {
    "EP_URL": "https://example.com/api"
  },
  "queues": {
    "producers": [{
      "queue": "queue-name",
      "binding": "DB_QUEUE"
    }],
    "consumers": [{
      "queue": "referral-engine",
      "max_batch_size": 5,
      "max_batch_timeout": 30,
      "max_concurrency": 10,
      "retry_delay": 120
    }]
  },
  "observability" : {
    "enabled": true,
    "head_sampling_rate": 1
  },
  "compatibility_date":"2024-11-11"
}

{
  "name": "my-worker",
  "workers_dev": true,
  "vars": {
    "EP_URL": "https://example.com/api"
  },
  "queues": {
    "producers": [{
      "queue": "queue-name",
      "binding": "DB_QUEUE"
    }],
    "consumers": [{
      "queue": "referral-engine",
      "max_batch_size": 5,
      "max_batch_timeout": 30,
      "max_concurrency": 10,
      "retry_delay": 120
    }]
  },
  "observability" : {
    "enabled": true,
    "head_sampling_rate": 1
  },
  "compatibility_date":"2024-11-11"
}

ambergristle•4d ago

thanks! full disclosure, i have limited experience w cloudflare

TheLegendVibesOP•4d ago

fetch function with post method is not working inside queue

ambergristle•4d ago

some of what you've got going on w Context doesn't totally make sense to me tho i think the place to start is by losing the any typing

TheLegendVibesOP•4d ago

any suggestion on improving this?

ambergristle•4d ago

i'm maybe a little extreme, but i only ever use any in generics and when possible/relevant, i use unknown instead so i'd start by: - always using c to stand for Context - always passing c as the first arg, if you're passing at all - typing as Context, if at all (#2 you can ease up on with time, but it's a good initial mental model, imo does that scan? when you use any, you essentially break typescript, and it's on you to fix it. these suggestions are meant to give us a clear picture of what your code is actually doing

TheLegendVibesOP•4d ago

understood, but I used unknown first, then used any to see if that makes any difference

ambergristle•4d ago

so this is a case where you definitely don't want to use either unknown or any c should always extend type Context why were you using any/unknown to begin w?

TheLegendVibesOP•4d ago

updated codes still not working

ambergristle•4d ago

this too? queue(batch: MessageBatch<QueueDataType>, c: Env) are you getting an error message?

TheLegendVibesOP•4d ago

no, I'm not getting any errors

ambergristle•4d ago

oh, i see, my bad. that's the queue fn signature in that case, queueHandler needs to take Env, not c/Context what's not working then?

TheLegendVibesOP•4d ago

fetch operations in queue handler

ambergristle•4d ago

ah, gotcha so what's happening?