better off not using DOs and just using waitUntil in a normal worker at that point

U

UnsmartOP•9/17/23, 10:42 PM

an alarm at the very least would retry like 3 times I think but you would need to save the value in the DO storage first

UUnsmart waitUntil does nothing in DOs and it can still fail

K

Kaique Nunes•9/17/23, 10:44 PM

actually I made some tests and it does. They limited the in-memory to 10 seconds. With waitUntil I made it live for 30 seconds. Not sure if it's possible to live more time, I will test

UUnsmart better off not using DOs and just using waitUntil in a normal worker at that poi...

K

Kaique Nunes•9/17/23, 10:45 PM

Yes, but I use the DO storage

U

UnsmartOP•9/17/23, 10:49 PM

The documentation got changed due to moving to its own section and some stuff seems to be forgotten/i cant find but as you can see from this message on the old docs: durable-objects

waitUntil in durable objects does nothing.

U

UnsmartOP•9/17/23, 10:53 PM

waitUntil/non awaited DO promises also has 0 guarentees of the promise actually completing and it does not retry any operations so there is a chance you return a success to the user but it does not succeed

K

Kaique Nunes•9/17/23, 10:57 PM

Not sure what is happening, but I created a simple counter that I can already persist in-memory for at least 2 minutes when using waintUntil with setTimeout. I am not persisting on storage. After 2 minutes it is cleaned.

U

UnsmartOP•9/17/23, 10:59 PM

any requests that happen to the DO will extend how long it lives and if there are any unfulfilled promises it might stay longer than the 10s documented time but having an unfulfilled promise wont guarantee it will eventually get fulfilled/rejected

K

Kaique Nunes•9/17/23, 11:03 PM

true! DO is awaiting independently of waitUntil.

KKaique Nunes actually I made some tests and it does. They limited the in-memory to 10 seconds...

K

Kaique Nunes•9/17/23, 11:05 PM

Well, but if it waits for the promise to complete, why would I use the alarm??

U

UnsmartOP•9/17/23, 11:06 PM

it waits but it doesnt guarantee it will actually finish... and it doesnt retry on any errors

K

Kaique Nunes•9/17/23, 11:07 PM

wow, great! thank you very much! I will do some tests

=

=HT=Chief•9/18/23, 8:51 PM

are there any plans to offer amqp support for remote integration with queues? I'd love to use this as an alternative to RabbitMQ.

CCharlie @htchief being able to consume from a Queue directly via API is on the roadmap. ...

=

=HT=Chief•9/18/23, 8:59 PM

right, but my understanding is that the protocol for that would be http, not amqp (Advanced Message Queuing Protocol)

=

=HT=Chief•9/18/23, 9:01 PM

the core difference to my understanding is that it's essentially single-socket-per-connection based, meaning that you can reduce the amount of requests / responses between endpoints

=

=HT=Chief•9/18/23, 9:02 PM

but the fact that there's already libraries across multiple languages which support the protocol is also nice

CCharlie Walk me through what you’re looking for here?

=

=HT=Chief•9/19/23, 12:45 AM

Apologies for the late response. From my understanding, the product that you're offering is very similar in functionality to RabbitMQ, which is a product that I use a lot. I would love to be able to use your product (when it is ready for this kind of testing) as a drop-in replacement - especially in the scenario of using an edge worker to enqueue messages to the queue for asynchronous processing.

U

UnsmartOP•9/19/23, 12:47 AM

Its meant to replicate AWS SQS rather than RabbitMQ technically. What specifically from RabbitMQ do you want? Not sure what all it offers

=

=HT=Chief•9/19/23, 12:48 AM

i'm not as familiar with SQS, but from a quick search, the functional difference between SQS and RabbitMQ is the hosting

=

=HT=Chief•9/19/23, 12:50 AM

otherwise they're similar products. The big appeal of RabbitMQ for me is the amqp protocol which is essentially socket based, so a new connection doesn't need to be made every time messages need to be fetched by the consumer (or the injestor)

U

UnsmartOP•9/19/23, 12:51 AM

afaik the only thing AWS SQS has over cloudflare queues is FIFO (first in first out) and HTTP api. SQS doesnt have any websockets afaik only http

=

=HT=Chief•9/19/23, 12:51 AM

and support for the amqp protocol would mean that i could use an existing library instead of having to write a whole new one (even if it is provided by as SDK)

=

=HT=Chief•9/19/23, 12:52 AM

ah, yes, RabbitMQ is FIFO out of the box. Also, i don't care for "websockets" - just a socket connection

=

=HT=Chief•9/19/23, 1:02 AM

here's a quick illustration of what i'm currently dealing with. What i'd like (in order to reduce the # of "subrequests") is to be able to route from a worker into a queue, and then to be able to consume the queue from my applications which are not workers (and are too complicated to be re-built in that environment)

=

=HT=Chief•9/19/23, 1:05 AM

again - as a "make a wish" for this product, i'd love it if i could use amqp as the protocol, not just HTTP. Here's more information about it:

https://www.amqp.org/

The rest of the product seems like it has a lot of potential and i'm looking forward to seeing how i can utilize it.

M

MetalGear•9/19/23, 12:52 PM

Is there a way to consume an event after a 10 minute delay instead of right away. I’m communicating with an iot device api and sometimes the third party is unreliable or down for couple hours so I have to retry every 30 min

CCharlie Not necessarily a solution to this problem, but have you looked into our PubSub ...

M

MetalGear•9/20/23, 12:33 AM

I actually have not because I thought queue was the reccomeneded solution for simply processing events. I just need a way to send a command to a iot device. If the consumer worker fails sending the fetch api call then I just want it to retry. Do you think I should look at it ? If so why? Thank you. I dont really understand the major difference between queues and pub/sub besides that pub sub has topics and subscribers and queues is just a linear stack that has events in order waiting for them to be consumed

==HT=Chief ah, yes, RabbitMQ is FIFO out of the box. Also, i don't care for "websockets" - ...

K

Kai•9/21/23, 8:58 AM

I just want to point out that HTTP requests beyond HTTP 1.1 are also kept alive if the implementation is decent. I would think that a future queue HTTP API will even support HTTP 3 which would allow you to pull a lot of messages over a single udp "connection". I highly doubt you'll have performance problems that aren't solved by a better HTTP implementation.

CCharlie Both can be great choices, depends on what your needs are. PubSub is based on th...

M

MetalGear•9/21/23, 1:32 PM

Is pub sub like queues in the sense that it’s guaranteed delivery at least once and rarely more then once?

Z

zegevlier•9/21/23, 1:52 PM

No, as per https://developers.cloudflare.com/pub-sub/platform/mqtt-compatibility/, pubsub currently only supports at most once delivery, not at least once.

KKai I just want to point out that HTTP requests beyond HTTP 1.1 are also kept alive ...

=

=HT=Chief•9/21/23, 6:28 PM

That is very interesting (i learned something new - thank you), but at the end of the day doesn't really answer on the ability to use pre-existing SDK's / libraries to integrate with queues. Its the difference between being able to use queues as a drop-in replacement vs having to build a completely new integration to work with it.

Zzegevlier No, as per <https://developers.cloudflare.com/pub-sub/platform/mqtt-compatibilit...

M

MetalGear•9/22/23, 1:36 AM

what i really dont understand is how are developers okay with at most once. Like who is okay with missing an event

S

Saifulapm•9/22/23, 8:10 AM

Can we use service binding in queue ? If yes, then what request should pass in fetch ??

O

Old Man Maker•9/22/23, 10:44 AM

hi there - I've got a queue. Got a producer and a consumer. In the dashboard, create a message - says successfully delivered! - I go above it and click on List messages - ZERO messages in the queue. Am I missing something here?

J

jahir9991•9/22/23, 10:57 AM

can i create a producer or consumer in a pages project like sveltekit?

O

Old Man Maker•9/22/23, 11:00 AM

Log message from my producer:

 {
      "message": [
        "Queue is dead=>",
        null
      ],
      "level": "log",
      "timestamp": 1695380365584
    },

 {
      "message": [
        "Queue is dead=>",
        null
      ],
      "level": "log",
      "timestamp": 1695380365584
    },

 {
      "message": [
        "Queue is dead=>",
        null
      ],
      "level": "log",
      "timestamp": 1695380365584
    },

 {
      "message": [
        "Queue is dead=>",
        null
      ],
      "level": "log",
      "timestamp": 1695380365584
    },

O

Old Man Maker•9/22/23, 11:03 AM

Checked Cloudflare Status - it's either lying or your service link is broken. Either way, I certainly can't push messages into my Queue. Can someone please update

Jjahir9991 can i create a producer or consumer in a pages project like sveltekit?

F

forslunds•9/22/23, 11:25 AM

Pages can only send (produce) not consume for the moment. Need worker for that

O

Old Man Maker•9/22/23, 11:36 AM

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...

M

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.

M

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...

O

Old Man Maker•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?

O

Old Man Maker•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.

N

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

[[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.

better off not using DOs and just using waitUntil in a normal worker at that point

Similar Threads

Similar Threads

Similar Threads