Multiple Producers on same Queue

I’m not an expert on miniflare but I’ll look into how to make sure multiple local workers are bound to the same Queue.

15 Replies

Following up on this. I played around with the mounting functionality described here: https://miniflare.dev/core/mount but wasn't able to get something to work. I'm following up with the author of Miniflare (he works at Cloudflare now) and maybe he'll have some ideas. If multiple workers writing to the same queue in local dev mode is a requested feature we can definitely prioritize that as part of our push towards GA.

🔌 Multiple Workers · Miniflare

Fun, full-featured, fully-local simulator for Cloudflare Workers

OilyLimeOP•2y ago

@charlieburnett for visibility 🙂

Ben Cox•2y ago

I'm with the OP on this one - just want to be able to test both my producer and my consumer, and don't need multiple producers.

OilyLimeOP•2y ago

To be clear, that's totally doable right now. You have to be using module Worker syntax (https://developers.cloudflare.com/workers/learning/migrating-to-module-workers/) and define both the fetch and queue entrypoints on the same script.

Migrating to module Workers · Cloudflare Workers docs

This guide will show you how to migrate your Workers from the Service Worker syntax to the new Module Worker format.

Ben Cox•2y ago

I see - so that would enable local dev testing for both producer and consumer? In production I was assuming these would need to be separate workers, but perhaps it doesn't actually matter

OilyLimeOP•2y ago

It's not as versatile as having two separate wrangler dev --local instances automagically bind to the same queue, but it does give you the ability to emulate a full queue workflow locally. You're free to do as you please, and it can totally be the same worker. Personally I do have an instance of the same Worker script being used as a producer and consumer. You can think of it as whether you want separate Docker images for an application that talks to SQS or not.

Ben Cox•2y ago

Yep - that works like a champ! Thanks @oilylime!

Jeff•2y ago

https://discord.com/channels/595317990191398933/891052295410835476/1094412946328789003 This may work as well, I can’t try until tomorrow

OilyLimeOP•2y ago

Following up here, got some help from our resident Miniflare author, and just tested it personally, here's something that works! Multiple producers, single consumer. index.js:

import { Miniflare } from "miniflare";

const mf = new Miniflare({
  mounts: {
    first: {
      // https://github.com/cloudflare/miniflare/blob/1a2cc368bfbe4d117e781ed4e74309c8da0c10d1/packages/queues/src/plugin.ts#L16-L19
      queueBindings: [{ name: "MY_QUEUE", queueName: "my-queue" }],
      routes: ["*/first"],
      modules: true,
      script: `export default {
        async fetch(request, env, ctx) {
          await env.MY_QUEUE.send("hello");
          return new Response("ok");
        }
      }`,
    },
    second: {
      // https://github.com/cloudflare/miniflare/blob/1a2cc368bfbe4d117e781ed4e74309c8da0c10d1/packages/queues/src/plugin.ts#L16-L19
      queueBindings: [{ name: "MY_QUEUE", queueName: "my-queue" }],
      routes: ["*/second"],
      modules: true,
      script: `export default {
        async fetch(request, env, ctx) {
          await env.MY_QUEUE.send("hello");
          return new Response("ok");
        }
      }`,
    },
    consumer: {
      // https://github.com/cloudflare/miniflare/blob/1a2cc368bfbe4d117e781ed4e74309c8da0c10d1/packages/queues/src/plugin.ts#L21-L27
      queueConsumers: [{ queueName: "my-queue" }],
      modules: true,
      script: `export default {
        async queue(messages, env, ctx) {
          console.log(messages);
        }
      }`,
    },
  },
});

await mf.startServer();

import { Miniflare } from "miniflare";

const mf = new Miniflare({
  mounts: {
    first: {
      // https://github.com/cloudflare/miniflare/blob/1a2cc368bfbe4d117e781ed4e74309c8da0c10d1/packages/queues/src/plugin.ts#L16-L19
      queueBindings: [{ name: "MY_QUEUE", queueName: "my-queue" }],
      routes: ["*/first"],
      modules: true,
      script: `export default {
        async fetch(request, env, ctx) {
          await env.MY_QUEUE.send("hello");
          return new Response("ok");
        }
      }`,
    },
    second: {
      // https://github.com/cloudflare/miniflare/blob/1a2cc368bfbe4d117e781ed4e74309c8da0c10d1/packages/queues/src/plugin.ts#L16-L19
      queueBindings: [{ name: "MY_QUEUE", queueName: "my-queue" }],
      routes: ["*/second"],
      modules: true,
      script: `export default {
        async fetch(request, env, ctx) {
          await env.MY_QUEUE.send("hello");
          return new Response("ok");
        }
      }`,
    },
    consumer: {
      // https://github.com/cloudflare/miniflare/blob/1a2cc368bfbe4d117e781ed4e74309c8da0c10d1/packages/queues/src/plugin.ts#L21-L27
      queueConsumers: [{ queueName: "my-queue" }],
      modules: true,
      script: `export default {
        async queue(messages, env, ctx) {
          console.log(messages);
        }
      }`,
    },
  },
});

await mf.startServer();

node --experimental-vm-modules index.js
MessageBatch {
  queue: 'my-queue',
  messages: [
    Message {
      body: 'hello',
      id: 'my-queue-0',
      timestamp: 2023-04-11T22:05:51.563Z
    }
  ]
}
MessageBatch {
  queue: 'my-queue',
  messages: [
    Message {
      body: 'hello',
      id: 'my-queue-1',
      timestamp: 2023-04-11T22:05:54.659Z
    }
  ]
}

node --experimental-vm-modules index.js
MessageBatch {
  queue: 'my-queue',
  messages: [
    Message {
      body: 'hello',
      id: 'my-queue-0',
      timestamp: 2023-04-11T22:05:51.563Z
    }
  ]
}
MessageBatch {
  queue: 'my-queue',
  messages: [
    Message {
      body: 'hello',
      id: 'my-queue-1',
      timestamp: 2023-04-11T22:05:54.659Z
    }
  ]
}

❯ curl 127.0.0.1:8787/second
ok%
❯ curl 127.0.0.1:8787/first
ok%

❯ curl 127.0.0.1:8787/second
ok%
❯ curl 127.0.0.1:8787/first
ok%

OilyLimeOP•2y ago

In the above example you'd want to swap out script from the mounts to the source of your worker scripts. Here's the miniflare API for reference: https://miniflare.dev/get-started/api

🧰 Using the API · Miniflare

Fun, full-featured, fully-local simulator for Cloudflare Workers

Jeff•2y ago

Gotcha. So the goal is a file that uses miniflare directly and can run all of my monorepo at once Thanks for going so deep on this guys

OilyLimeOP•2y ago

Yeah, by invoking the Miniflare API in js directly you have a good amount of control. It was a learning experience for me too, and a valuable one.

Unknown User•2y ago

Message Not Public

Jeff•2y ago

Yeah, you need to run it through esbuild first. The example someone sent me used esbuild directly, although there may be a way to do it with miniflare directly

Unknown User•2y ago

Message Not Public

Gaming

Programming

Multiple Producers on same Queue