Merge pull request #133 from syumai/split-producer-send-message-funcs

split queues.Producer's Send funcs and BatchMessage constructors

Author: syumai

_examples/queues/Makefile

@@ -1,6 +1,6 @@
 .PHONY: dev
 dev:
-	npx wrangler dev --port 8787
+	wrangler dev
 
 .PHONY: build
 build:
@@ -9,4 +9,4 @@ build:
 
 .PHONY: deploy
 deploy:
-	npx wrangler deploy
+	wrangler deploy

_examples/queues/main.go

@@ -23,6 +23,7 @@ func main() {
 	http.HandleFunc("/", handleProduce)
 	workers.Serve(nil)
 }
+
 func handleProduce(w http.ResponseWriter, req *http.Request) {
 	if req.URL.Path != "/" {
 		w.WriteHeader(http.StatusNotFound)
@@ -48,7 +49,7 @@ func handleProduce(w http.ResponseWriter, req *http.Request) {
 		err = produceText(q, req)
 	case "application/json":
 		log.Println("Handling json content type")
-		err = produceJson(q, req)
+		err = produceJSON(q, req)
 	default:
 		log.Println("Handling bytes content type")
 		err = produceBytes(q, req)
@@ -68,38 +69,33 @@ func produceText(q *queues.Producer, req *http.Request) error {
 	if err != nil {
 		return fmt.Errorf("failed to read request body: %w", err)
 	}
-	if len(content) == 0 {
-		return fmt.Errorf("empty request body")
-	}
-
-	// text content type supports string and []byte messages
-	if err := q.Send(content, queues.WithContentType(queues.QueueContentTypeText)); err != nil {
+	// text content type supports string
+	if err := q.SendText(string(content)); err != nil {
 		return fmt.Errorf("failed to send message: %w", err)
 	}
-
 	return nil
 }
 
-func produceJson(q *queues.Producer, req *http.Request) error {
+func produceJSON(q *queues.Producer, req *http.Request) error {
 	var data any
 	if err := json.NewDecoder(req.Body).Decode(&data); err != nil {
 		return fmt.Errorf("failed to read request body: %w", err)
 	}
-
-	// json content type is default and therefore can be omitted
 	// json content type supports messages of types that can be serialized to json
-	if err := q.Send(data); err != nil {
+	if err := q.SendJSON(data); err != nil {
 		return fmt.Errorf("failed to send message: %w", err)
 	}
-
 	return nil
 }
 
 func produceBytes(q *queues.Producer, req *http.Request) error {
-	// bytes content type support messages of type []byte, string, and io.Reader
-	if err := q.Send(req.Body, queues.WithContentType(queues.QueueContentTypeBytes)); err != nil {
+	content, err := io.ReadAll(req.Body)
+	if err != nil {
+		return fmt.Errorf("failed to read request body: %w", err)
+	}
+	// bytes content type support messages of type []byte
+	if err := q.SendBytes(content); err != nil {
 		return fmt.Errorf("failed to send message: %w", err)
 	}
-
 	return nil
 }

cloudflare/queues/batchmessage.go

@@ -0,0 +1,50 @@
+package queues
+
+import (
+	"syscall/js"
+
+	"github.com/syumai/workers/internal/jsutil"
+)
+
+type BatchMessage struct {
+	body    js.Value
+	options *sendOptions
+}
+
+// NewTextBatchMessage creates a single text message to be batched before sending to a queue.
+func NewTextBatchMessage(content string, opts ...SendOption) *BatchMessage {
+	return newBatchMessage(js.ValueOf(content), contentTypeText, opts...)
+}
+
+// NewBytesBatchMessage creates a single byte array message to be batched before sending to a queue.
+func NewBytesBatchMessage(content []byte, opts ...SendOption) *BatchMessage {
+	return newBatchMessage(js.ValueOf(content), contentTypeBytes, opts...)
+}
+
+// NewJSONBatchMessage creates a single JSON message to be batched before sending to a queue.
+func NewJSONBatchMessage(content any, opts ...SendOption) *BatchMessage {
+	return newBatchMessage(js.ValueOf(content), contentTypeJSON, opts...)
+}
+
+// NewV8BatchMessage creates a single raw JS value message to be batched before sending to a queue.
+func NewV8BatchMessage(content js.Value, opts ...SendOption) *BatchMessage {
+	return newBatchMessage(content, contentTypeV8, opts...)
+}
+
+// newBatchMessage creates a single message to be batched before sending to a queue.
+func newBatchMessage(body js.Value, contentType contentType, opts ...SendOption) *BatchMessage {
+	options := sendOptions{
+		ContentType: contentType,
+	}
+	for _, opt := range opts {
+		opt(&options)
+	}
+	return &BatchMessage{body: body, options: &options}
+}
+
+func (m *BatchMessage) toJS() js.Value {
+	obj := jsutil.NewObject()
+	obj.Set("body", m.body)
+	obj.Set("options", m.options.toJS())
+	return obj
+}

cloudflare/queues/batchsendoptions.go

@@ -0,0 +1,36 @@
+package queues
+
+import (
+	"syscall/js"
+	"time"
+
+	"github.com/syumai/workers/internal/jsutil"
+)
+
+type batchSendOptions struct {
+	// DelaySeconds - The number of seconds to delay the message.
+	// Default is 0
+	DelaySeconds int
+}
+
+func (o *batchSendOptions) toJS() js.Value {
+	if o == nil {
+		return js.Undefined()
+	}
+
+	obj := jsutil.NewObject()
+	if o.DelaySeconds != 0 {
+		obj.Set("delaySeconds", o.DelaySeconds)
+	}
+
+	return obj
+}
+
+type BatchSendOption func(*batchSendOptions)
+
+// WithBatchDelaySeconds changes the number of seconds to delay the message.
+func WithBatchDelaySeconds(d time.Duration) BatchSendOption {
+	return func(o *batchSendOptions) {
+		o.DelaySeconds = int(d.Seconds())
+	}
+}