Implement Queues Consumer handler

[syumai]

• Implement Queues Consumer handler and add to Module Worker.
  • https://developers.cloudflare.com/queues/platform/javascript-apis/

[eliezedeck]

Please 🙏

[syumai]

I'll try to implement this in next few weeks.

[eliezedeck]

I'll try to implement this in next few weeks.

Thanks in advance. By the way, would you mind sharing the concrete steps that will be involved in doing the implementation? I'm curious to know how things work under the hood, who knows, I might even be able to contribute?

[syumai]

It will be like the following steps.

• producer side
  • add Wrapper package of Queue class (send / sendBatch)
    • https://developers.cloudflare.com/queues/reference/javascript-apis/#producer
• consumer side (almost same as Add Scheduled worker #45)
  1. add queue handler to workers-assets-gen command's shim.mjs
  2. add function to listen queue handler (like cron.go of Add Scheduled worker #45).
• add example

PRs are welcome!

[meandnano]

I've started working on this (#125) since the I need the Queues support in my worker. Though there's a problem that I'd like to discuss as my experience with workers and js in general is not that great.

In general, producer's side of API is clear and mostly works as expected. One important feature of producer's api is an option to specify the content type of each message produced. This is handled in my PR.

The questions I have are concerning the consumer's API.
What I don't really understand is how to handle the expectedly most common case of incoming messages - JSON objects. On the receiving side in Go we're dealing with js.Value type, for which I don't see a straightforward way of mapping (or "unmarshaling") to any or map[string]any.
As an option I was thinking of explicitly calling JS's JSON.stringify() over the incoming messages bodies in order to later json.Unmarhal() them in go code, but this looks like an ugly waste of resources.

Any advices on how to achieve this?

[syumai]

Thank you @meandnano!
#125 looks so nice. I'm looking forward to seeing the PR opened.

As an option I was thinking of explicitly calling JS's JSON.stringify() over the incoming messages bodies in order to later json.Unmarhal() them in go code, but this looks like an ugly waste of resources.

I think we should provide a way to access to the raw value of the message.

Almost all the API can be the same as the Queue Consumer API (MessageBatch type, Message type ... etc).
The only issue we need to address is how to handle Message.body.

I think we need to introduce a type for MessageBody, with methods like String(), Raw(), etc.

• String() (string, error) returns the string value of the body only if the JavaScript value type of the body is string. Otherwise, it returns an error.
• Raw() js.Value returns the raw representation of the body.

Then, Raw() value can be converted to structs using a 3rd party library such as go-jsutil.
Here's my implementation of go-jsutil/encoding.MarshalJSONValue / UnmarshalJSONValue.
It can be used like:

import jsencoding "github.com/syumai/go-jsutil/encoding"

type QueueMsg struct {
	Type string
	Value int
}

func Consume(batch queues.MessageBatch) {
	for _, msg := range batch.Messages {
		var queueMsg QueueMsg
		err := jsencoding.UnmarshalJSONValue(msg.Raw(), &queueMsg)
		if err != nil { /* handling error */ }
		// Now, queueMsg is available
	}
}

UnmarshalJSONValue is not efficient, however by exposing the raw representation of the value, users may choose to use a more efficient conversion method.

I don't think functions like UnmarshalJSONValue should be provided as a feature of syumai/workers.

I hope this answer helps you with your implementation.

[meandnano]

Thanks for the input @syumai. My current implementation of consumer API does exactly that - exposes the raw js.Value:

type ConsumerMessage struct {
	// instance - The underlying instance of the JS message object passed by the cloudflare
	instance js.Value

	Id        string
	Timestamp time.Time
	Body      js.Value
	Attempts  int
}

IMO we don't need Raw() js.Value since Body is exported. Also, given that jsutil is internal, adding body conversion functions is indeed a good idea.

Essentially, I have it all working locally and now am only focused on the tests. Will push updates soon enough