so messages can be pushed into the queue from dashboard again. in my consumer Worker, I want to rou

so messages can be pushed into the queue from dashboard again.

in my consumer Worker, I want to route the queue message back to a client using a websocket. I'm not having much luck.

async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
        
        try {
            
            let messages = JSON.stringify(batch.messages);
            console.log(`consumed from our queue: ${messages}`);

            if(serverSocket !== undefined){
                console.log('ServerSocket=>', serverSocket) ;
                serverSocket.send(messages);

            }else{
                console.log('ServerSocket Undefined=>', serverSocket) ;

                console.log('Sending from Running Sock=>', serverSocket) ;
                env.runningSock.send(messages);
            }

        } catch (error) {
            console.log('Error=>',error);
        }
      },

async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
        
        try {
            
            let messages = JSON.stringify(batch.messages);
            console.log(`consumed from our queue: ${messages}`);

            if(serverSocket !== undefined){
                console.log('ServerSocket=>', serverSocket) ;
                serverSocket.send(messages);

            }else{
                console.log('ServerSocket Undefined=>', serverSocket) ;

                console.log('Sending from Running Sock=>', serverSocket) ;
                env.runningSock.send(messages);
            }

        } catch (error) {
            console.log('Error=>',error);
        }
      },

async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
        
        try {
            
            let messages = JSON.stringify(batch.messages);
            console.log(`consumed from our queue: ${messages}`);

            if(serverSocket !== undefined){
                console.log('ServerSocket=>', serverSocket) ;
                serverSocket.send(messages);

            }else{
                console.log('ServerSocket Undefined=>', serverSocket) ;

                console.log('Sending from Running Sock=>', serverSocket) ;
                env.runningSock.send(messages);
            }

        } catch (error) {
            console.log('Error=>',error);
        }
      },

async queue(batch: MessageBatch<any>, env: Env): Promise<void> {
        
        try {
            
            let messages = JSON.stringify(batch.messages);
            console.log(`consumed from our queue: ${messages}`);

            if(serverSocket !== undefined){
                console.log('ServerSocket=>', serverSocket) ;
                serverSocket.send(messages);

            }else{
                console.log('ServerSocket Undefined=>', serverSocket) ;

                console.log('Sending from Running Sock=>', serverSocket) ;
                env.runningSock.send(messages);
            }

        } catch (error) {
            console.log('Error=>',error);
        }
      },

My socket connects correctly and produced test messages etc - but as soon as the queue has a message, the socket is null. I've tried putting the server object from the WebsocketPair into a variable both local and in the env object. Always null.

How can I store the active websocket from the worker, so the queue message can be send ?

OOld Man Maker so messages can be pushed into the queue from dashboard again. in my consumer W...

Matt Silverlock•9/22/23, 12:48 PM

A Queue consumer would need to dial out - a consumer runs separately from any

fetch()

fetch()

handler. It’s null because I suspect you’re trying to pass the socket (somehow?) across handlers: you can’t.

Matt Silverlock•9/22/23, 12:50 PM

More context on what you’re trying to build (not the tech; the overall problem/architecture) would be helpful but I suspect you may be better served by Durable Objects if you are trying to build something more “real time”. Write to a DO, use an alarm() to wake it, push messages to connected clients over WS.

MMatt Silverlock More context on what you’re trying to build (not the tech; the overall problem/a...

Old Man MakerOP•9/22/23, 1:12 PM

Thanks @elithrar for the response - I'm very new to queues. I have client that calls worker A which is the queue producer. the same client connects to worker B which is a websocket and the queue consumer. The reason for this is that worker A triggers a few things, but I need a valued pushed back to the client quickly while the rest of the actions are occuring. I thought I could use the queue to make this work. I'm not at all married to this solution, but I need that value passed back to the client, before the rest of the response gets returned from worker A if that makes sense?

Old Man MakerOP•9/22/23, 1:16 PM

I think your suggestion of a DO is right on the money. I need the data being returned to contain an identifier that a specific client can read. I'll give that a shot.
Thanks for explaining the queue is in another process. That makes sense and clarifies things for me.

nuno•9/22/23, 3:01 PM

Hey guys! I'm new to queues, and now I'm having an error from the send method but the error is super generic,

Internal Server Error

Internal Server Error

.

Do you guys know what could be?

This is my wrangler.toml

[[queues.producers]]
 queue = "pdx-dev-stripe-queue"
 binding = "STRIPE_QUEUE"

[[queues.consumers]]
  queue = "pdx-dev-stripe-queue"
  max_batch_size = 1
  max_batch_timeout = 30

[[queues.producers]]
 queue = "pdx-dev-stripe-queue"
 binding = "STRIPE_QUEUE"

[[queues.consumers]]
  queue = "pdx-dev-stripe-queue"
  max_batch_size = 1
  max_batch_timeout = 30

Thanks in advance.

Nnuno Hey guys! I'm new to queues, and now I'm having an error from the send method bu...

Old Man MakerOP•9/22/23, 3:12 PM

Hi - I'm new myself, so this may be BS - but I think you need to define 2 workers, one that is the producer and one that is the consumer.

OOld Man Maker Hi - I'm new myself, so this may be BS - but I think you need to define 2 worker...

nuno•9/22/23, 3:17 PM

I found this in the documentation
https://developers.cloudflare.com/queues/learning/local-development/#known-issues

Known Issues
Wrangler does not yet support running separate producer and consumer Workers bound to the same Queue locally. To develop locally with Queues, you can temporarily put your consumer’s queue() handler in the same Worker as your producer, so the same Worker acts as both a producer and consumer.

Wrangler also does not yet support wrangler dev --remote.

So, I assume is it possible to run both producer and consumer in the same Worker.

nuno•9/22/23, 3:19 PM

For some reason, the Cloudflare Dashboard, shows that the queue has no producer worker linked

Nnuno For some reason, the Cloudflare Dashboard, shows that the queue has no producer ...

Old Man MakerOP•9/22/23, 3:30 PM

Hey thanks for that clarification. I created two for deployment because I couldn't get local to work (which that now explains it) - PITA to have to then break up for deployment.

OOld Man Maker Hey thanks for that clarification. I created two for deployment because I couldn...

nuno•9/22/23, 3:33 PM

Still dont know how to link the producer to the queue

Old Man MakerOP•9/22/23, 3:35 PM

I'm not sure if that is because you've got them in the same file? You can try and creaet another Worker and add the producer into that one to see if it links in the dashboard. I can confirm that mine is showing correectly.

OOld Man Maker I'm not sure if that is because you've got them in the same file? You can try an...

nuno•9/22/23, 3:45 PM

Well, I got it to work. It was the beter-sqlite package facePalm

Nnuno Well, I got it to work. It was the beter-sqlite package <:facePalm:7403809801158...

Old Man MakerOP•9/22/23, 3:46 PM

Nice

OOld Man Maker Nice

nuno•9/22/23, 3:46 PM

Thanks for your help sir

Old Man MakerOP•9/22/23, 3:47 PM

I was exactly no-help at all

Thanks for sharing what you learnt. That's about the only way to figure some of this out I think

Adi//•9/24/23, 5:24 AM

Hi folks is queues beta free to use?

Adi//•9/24/23, 5:24 AM

Or I need to be on the workers paid plan for 5$ first?

Adi//•9/24/23, 5:25 AM

trying to understand what this means

Screenshot_2023-09-24_at_10.55.26_AM.png

Screenshot_2023-09-24_at_10.55.36_AM.png

AAdi//Or I need to be on the workers paid plan for 5$ first?

kian•9/24/23, 5:39 AM

Yep

Kkian Yep

Adi//•9/24/23, 6:15 AM

ok thanks!

nuno•9/26/23, 9:06 AM

Hey guys! I'm using the queue to handle Stripe Events (Webhooks), and I want to make a fetch to another API, but the instant the code reaches the

fetch

fetch

, it just skip / stops running. Yesterday I assumed that the

fetches

fetches

were disabled on the queue, but minutes ago I saw an example on the CF Docs where a

fetch

fetch

is made inside a queue.

https://developers.cloudflare.com/queues/examples/web-crawler-with-browser-rendering/#6-crawl-with-puppeteer

Can someone help me understand what could be the problem?
Thanks in advance.

Nnuno Hey guys! I'm using the queue to handle Stripe Events (Webhooks), and I want to ...

Matt Silverlock•9/26/23, 10:47 AM

Can you share your code? Are you correctly

await

await

ing the fetch? Do you have a

try-catch

try-catch

handler and are you seeing any errors logged?

MMatt Silverlock Can you share your code? Are you correctly `await` ing the fetch? Do you have a ...

nuno•9/26/23, 10:55 AM

Yes, I'm

await

await

ing the

fetch

fetch

and I also put inside a

try/catch

try/catch

, but no error is logged.
The last log I see is the

API URL

API URL

log, and then the code skips everything below the log

try {
    console.log('API URL :', `${env.STRAPI_API_URL}/c-orders`);
    const connection = await fetch(`${env.STRAPI_API_URL}/c-orders`, {
      headers: {
        Authorization: `Bearer ${env.STRAPI_API_KEY}`,
        'Content-Type': 'application/json',
      },
      method: 'POST',
      body: JSON.stringify({
        data: {
          planId,
          date: new Date(),
          deliveryDate: new Date(deliveryDate * 1000),
          status: 1,
          details,
          grossPrice: totalPrice,
          finalPrice: totalPrice,
          subscriptionId: subscription ? strapiSubscriptionId : null,
          userId,
          paymentId: paymentIntent,
          paymentStatus: subscription ? 'processing' : 'success',
          orderInfo: [],
          stripeInvoiceId: invoice,
        },
      }),
    });

    console.log('connection :', connection);
    const data = await connection.json();
    console.log('data :', JSON.stringify(data, null, 2));

    if (connection.status !== 200) wasOrderCreated = false;

    wasOrderCreated = true;
  } catch (err) {
    console.log('err :', err);
    return false;
  }

try {
    console.log('API URL :', `${env.STRAPI_API_URL}/c-orders`);
    const connection = await fetch(`${env.STRAPI_API_URL}/c-orders`, {
      headers: {
        Authorization: `Bearer ${env.STRAPI_API_KEY}`,
        'Content-Type': 'application/json',
      },
      method: 'POST',
      body: JSON.stringify({
        data: {
          planId,
          date: new Date(),
          deliveryDate: new Date(deliveryDate * 1000),
          status: 1,
          details,
          grossPrice: totalPrice,
          finalPrice: totalPrice,
          subscriptionId: subscription ? strapiSubscriptionId : null,
          userId,
          paymentId: paymentIntent,
          paymentStatus: subscription ? 'processing' : 'success',
          orderInfo: [],
          stripeInvoiceId: invoice,
        },
      }),
    });

    console.log('connection :', connection);
    const data = await connection.json();
    console.log('data :', JSON.stringify(data, null, 2));

    if (connection.status !== 200) wasOrderCreated = false;

    wasOrderCreated = true;
  } catch (err) {
    console.log('err :', err);
    return false;
  }

Altrius•9/26/23, 9:04 PM

ok, so I started using Queues yesterday, and whilst the workers SDK is great, the one thing which would be a perfect addition for me would be the

send

send

and

sendBatch

sendBatch

methods returning a single (or array of for batch) string(s) representing each message sent to the queue

those IDs do exist, because they get given to the handler to ack, but theyre not available to the sending worker, it just returns a

Promise<void>

Promise<void>

, is there any chance that this might come along, or even be an option to enable it in the command itself (eg

MY_QUEUE.send('my payload', {resolveId: true})

MY_QUEUE.send('my payload', {resolveId: true})

Im only even mentioning it because I am using Queues alongside an external RabbitMQ server, where I acknowledge a queue job is completed back to my own local server for the next set of tasks to be triggered

Adi//•9/27/23, 12:47 PM

In case of workers the workers die when the client disconnects,
What is the case with queues?
Do queue workers have any "wall time or limit" (assuming the other limits like cpu time etc are not exceeded)

CCharlie In what situation are you imaging a queue consumer worker getting disconnected f...

Adi//•9/27/23, 8:53 PM

Nothing really, my worker will connect with another api and operate over streamed responses (SSE events) just curious if i still need to worry about Anything

Unsmart•9/27/23, 9:48 PM

If the worker errors it wont delete the messages from the queue so they would get retried eventually

UUnsmart If the worker errors it wont delete the messages from the queue so they would ge...

Adi//•9/28/23, 5:41 AM

oh this is super cool then imo, trying to understand how this works,

are messages deleted only after a task is completed? or thats upto me

Adi//•9/28/23, 5:41 AM

Also I assume duplicacy is handled directly by cloudflare (one message = one worker instance?)

Adi//•9/28/23, 5:43 AM

can someone help me understand these 2?

Screenshot_2023-09-28_at_11.13.36_AM.png

Screenshot_2023-09-28_at_11.13.47_AM.png

CCharlie @Adi// messages are deleted in 3 different circumstances - When the message is e...

Adi//•9/28/23, 6:45 PM

Got it understood, this helps

Adi//•9/28/23, 6:46 PM

Also I assume that means workers have no runtime limits (wall time) until I am well under cpu limits?

gideonred•9/29/23, 2:48 PM

Hello, ex-Cloudflarian here

. Two questions:

Is there a timeline for Queues to be supported in Wrangler dev preview mode?
We currently have Queues enabled for our CF prod account, but not non-prod. Our infrastructure team has reached out, but I haven't heard back from my internal team. Any chance someone here is motived to help me get it enabled?

gideonred•9/29/23, 2:58 PM

Also, is there a direct HTTP POST I can do to a queue for testing purposes?

Matt Silverlock•9/29/23, 5:46 PM

Gideon!

See this note: https://developers.cloudflare.com/queues/learning/local-development/#known-issues
- we’re working on a fix for this but won’t be this month.
Email me your account ID. Queues requires a paid Workers plan - that’s it. What are you seeing that isn’t working if that is the case?
HTTP POST - there is a dash debugging UI for testing. Not yet a REST API you can use.

Local Development · Cloudflare Queues

Queues support local development workflows using Wrangler, the command-line interface for Workers. Wrangler runs the same version of Queues as …

gideonred•9/29/23, 8:19 PM

@Matt !

Thanks for 1 and 3.

For 2: we finally got a reply from our Cloudflare TAM and yes, they stated we need to add a payment method on our non-prod account. I'm working on us setting that up.

MetalGear•10/2/23, 3:42 PM

Hey guys,

really loving my solution using only cloudflare services for my backend api but im currently suffering from one problem I am hoping someone can guide me. I am using cloudflare queues to manage events that need to be processed say sending a command to a iot device. That iot device 3rd party api has been giving me issues sometimes where its unnavailable for 3 hours due to maintnanace on their end. So when my worker consumes that queue I need a way to keep retrying all the queue items every hour. until they succeed and then I can mark them as consumed.

MetalGear•10/2/23, 4:04 PM

Do you guys think i should just use upstash for now if I need those features?

switz•10/3/23, 8:07 PM

whats the best way to retry a message in the queue but on a delay?

switz•10/3/23, 8:08 PM

will retries happen immediately?

Sswitz whats the best way to retry a message in the queue but on a delay?

MetalGear•10/3/23, 10:22 PM

No good way to do that right now unfortunetely based on my limited experience. I needed a solution fast so i just temporarily moved to qstash from upstash

switz•10/3/23, 10:46 PM

alternatively, would be nice to consume a queue from an external application. that would solve the problem for me

switz•10/3/23, 10:46 PM

so I only consume items when my server is up and running

switz•10/3/23, 10:47 PM

maybe this: https://idearium.github.io/idearium-lib/docs/cloudflare-queues

@idearium/cloudflare-queues | Idearium Lib

Safely process Cloudflare Queues batches and messages.

try { console.log('API URL :', `${env.STRAPI_API_URL}/c-orders`); const connection = await fetch(`${env.STRAPI_API_URL}/c-orders`, { headers: { Authorization: `Bearer ${env.STRAPI_API_KEY}`, 'Content-Type': 'application/json', }, method: 'POST', body: JSON.stringify({ data: { planId, date: new Date(), deliveryDate: new Date(deliveryDate * 1000), status: 1, details, grossPrice: totalPrice, finalPrice: totalPrice, subscriptionId: subscription ? strapiSubscriptionId : null, userId, paymentId: paymentIntent, paymentStatus: subscription ? 'processing' : 'success', orderInfo: [], stripeInvoiceId: invoice, }, }), }); console.log('connection :', connection); const data = await connection.json(); console.log('data :', JSON.stringify(data, null, 2)); if (connection.status !== 200) wasOrderCreated = false; wasOrderCreated = true; } catch (err) { console.log('err :', err); return false; }

so messages can be pushed into the queue from dashboard again. in my consumer Worker, I want to rou

Similar Threads

Similar Threads

Similar Threads