Multiple Producers on same Queue

Emilio · 2023-04-07T22:15:27.978Z

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.

E

EmilioOP•4/10/23, 9:01 PM

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

E

EmilioOP•4/10/23, 9:09 PM

@charlieburnett for visibility

B

Ben Cox•4/10/23, 9:09 PM

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.

BBen Cox I'm with the OP on this one - just want to be able to test both my producer and ...

E

EmilioOP•4/10/23, 9:12 PM

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

fetch

and

queue

queue

queue

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.

B

Ben Cox•4/10/23, 9:14 PM

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

E

EmilioOP•4/10/23, 9:14 PM

It's not as versatile as having two separate

wrangler dev --local

wrangler dev --local

wrangler dev --local

wrangler dev --local instances automagically bind to the same queue, but it does give you the ability to emulate a full queue workflow locally.

BBen Cox I see - so that would enable local dev testing for both producer and consumer? I...

E

EmilioOP•4/10/23, 9:16 PM

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.

B

Ben Cox•4/10/23, 9:18 PM

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

J

Jeff•4/11/23, 3:56 PM

miniflare

J

Jeff•4/11/23, 3:56 PM

This may work as well, I can’t try until tomorrow

E

EmilioOP•4/11/23, 10:07 PM

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

index.js

index.js

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();

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
    }
  ]
}

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%

❯ 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%

E

EmilioOP•4/11/23, 10:15 PM

In the above example you'd want to swap out

script

script

script

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

J

Jeff•4/12/23, 2:53 PM

Gotcha. So the goal is a file that uses miniflare directly and can run all of my monorepo at once

J

Jeff•4/12/23, 2:54 PM

Thanks for going so deep on this guys

E

EmilioOP•4/12/23, 3:45 PM

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.

J

Jeff•4/26/23, 1:33 PM

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

Similar Threads

Similar Threads

Similar Threads