Merge pull request #773 from tobilg/feature/cf-queues

Cloudflare Queues endpoint implementation

Author: tidwall

go.mod

@@ -23,16 +23,16 @@ require (
 	github.com/tidwall/buntdb v1.2.9
 	github.com/tidwall/expr v0.13.0
 	github.com/tidwall/geojson v1.4.5
-	github.com/tidwall/gjson v1.14.3
+	github.com/tidwall/gjson v1.18.0
 	github.com/tidwall/hashmap v1.6.1
 	github.com/tidwall/limiter v0.4.0
 	github.com/tidwall/match v1.1.1
-	github.com/tidwall/pretty v1.2.0
+	github.com/tidwall/pretty v1.2.1
 	github.com/tidwall/redbench v0.1.0
 	github.com/tidwall/redcon v1.4.4
 	github.com/tidwall/resp v0.1.1
 	github.com/tidwall/rtree v1.9.2
-	github.com/tidwall/sjson v1.2.4
+	github.com/tidwall/sjson v1.2.5
 	github.com/tidwall/tinylru v1.2.1
 	github.com/xdg-go/scram v1.1.2
 	github.com/yuin/gopher-lua v1.1.0
@@ -62,6 +62,7 @@ require (
 	github.com/Azure/go-autorest/tracing v0.6.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/cloudflare/cloudflare-go/v4 v4.6.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/devigned/tab v0.1.1 // indirect
 	github.com/eapache/go-resiliency v1.3.0 // indirect

go.sum

@@ -112,6 +112,8 @@ github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWR
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
+github.com/cloudflare/cloudflare-go/v4 v4.6.0 h1:ZaWwXjHFR5NoY8UEf4QFY0g3KTi72kqqEXpajV610/o=
+github.com/cloudflare/cloudflare-go/v4 v4.6.0/go.mod h1:XcYpLe7Mf6FN87kXzEWVnJ6z+vskW/k6eUqgqfhFE9k=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
@@ -377,8 +379,11 @@ github.com/tidwall/geoindex v1.7.0/go.mod h1:rvVVNEFfkJVWGUdEfU8QaoOg/9zFX0h9ofW
 github.com/tidwall/geojson v1.4.5 h1:BFVb5Pr7WZJMqFXy1LVudt5hPEWR3g4uhjk5Ezc3GzA=
 github.com/tidwall/geojson v1.4.5/go.mod h1:1cn3UWfSYCJOq53NZoQ9rirdw89+DM0vw+ZOAVvuReg=
 github.com/tidwall/gjson v1.12.1/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/gjson v1.14.3 h1:9jvXn7olKEHU1S9vwoMGliaT8jq1vJ7IH/n9zD9Dnlw=
 github.com/tidwall/gjson v1.14.3/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/gjson v1.18.0 h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY=
+github.com/tidwall/gjson v1.18.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
 github.com/tidwall/grect v0.1.4 h1:dA3oIgNgWdSspFzn1kS4S/RDpZFLrIxAZOdJKjYapOg=
 github.com/tidwall/grect v0.1.4/go.mod h1:9FBsaYRaR0Tcy4UwefBX/UDcDcDy9V5jUcxHzv2jd5Q=
 github.com/tidwall/hashmap v1.6.1 h1:FIAHjKwcyOo1Y3/orsQO08floKhInbEX2VQv7CQRNuw=
@@ -391,6 +396,8 @@ github.com/tidwall/match v1.1.1 h1:+Ho715JplO36QYgwN9PGYNhgZvoUSc9X2c80KVTi+GA=
 github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
 github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
+github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tidwall/redbench v0.1.0 h1:UZYUMhwMMObQRq5xU4SA3lmlJRztXzqtushDii+AmPo=
 github.com/tidwall/redbench v0.1.0/go.mod h1:zxcRGCq/JcqV48YjK9WxBNJL7JSpMzbLXaHvMcnanKQ=
 github.com/tidwall/redcon v1.4.4 h1:N3ZwZx6n5dqNxB3cfmj9D/8zNboFia5FAv1wt+azwyU=
@@ -404,6 +411,8 @@ github.com/tidwall/rtree v1.9.2 h1:6HiSU/bf4a7l2smEC+fEum/WloHMFCIQKWHjahm0Do8=
 github.com/tidwall/rtree v1.9.2/go.mod h1:iDJQ9NBRtbfKkzZu02za+mIlaP+bjYPnunbSNidpbCQ=
 github.com/tidwall/sjson v1.2.4 h1:cuiLzLnaMeBhRmEv00Lpk3tkYrcxpmbU81tAY4Dw0tc=
 github.com/tidwall/sjson v1.2.4/go.mod h1:098SZ494YoMWPmMO6ct4dcFnqxwj9r/gF0Etp19pSNM=
+github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY=
+github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28=
 github.com/tidwall/tinylru v1.2.1 h1:VgBr72c2IEr+V+pCdkPZUwiQ0KJknnWIYbhxAVkYfQk=
 github.com/tidwall/tinylru v1.2.1/go.mod h1:9bQnEduwB6inr2Y7AkBP7JPgCkyrhTV/ZpX0oOOpBI4=
 github.com/tidwall/tinyqueue v0.1.1 h1:SpNEvEggbpyN5DIReaJ2/1ndroY8iyEGxPYxoSaymYE=

internal/endpoint/cfqueue.go

@@ -0,0 +1,92 @@
+package endpoint
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/cloudflare/cloudflare-go/v4"
+	"github.com/cloudflare/cloudflare-go/v4/option"
+	"github.com/cloudflare/cloudflare-go/v4/queues"
+)
+
+const cfqueueExpiresAfter = time.Second * 30
+
+// CFQueueConn is an endpoint connection
+type CFQueueConn struct {
+	mu     sync.Mutex
+	ep     Endpoint
+	client *cloudflare.Client
+	ex     bool
+	t      time.Time
+}
+
+// Expired returns true if the connection has expired
+func (conn *CFQueueConn) Expired() bool {
+	conn.mu.Lock()
+	defer conn.mu.Unlock()
+	if !conn.ex {
+		if time.Since(conn.t) > cfqueueExpiresAfter {
+			conn.close()
+			conn.ex = true
+		}
+	}
+	return conn.ex
+}
+
+// ExpireNow forces the connection to expire
+func (conn *CFQueueConn) ExpireNow() {
+	conn.mu.Lock()
+	defer conn.mu.Unlock()
+	conn.close()
+	conn.ex = true
+}
+
+func (conn *CFQueueConn) close() {
+	if conn.client != nil {
+		conn.client = nil
+	}
+}
+
+// Send sends a message
+func (conn *CFQueueConn) Send(msg string) error {
+	conn.mu.Lock()
+	defer conn.mu.Unlock()
+
+	if conn.ex {
+		return errExpired
+	}
+	conn.t = time.Now()
+
+	// Initialize client if not already done
+	if conn.client == nil {
+		conn.client = cloudflare.NewClient(
+			option.WithAPIToken(conn.ep.CFQueue.APIToken),
+		)
+	}
+
+	// Push message to CF Queue
+	_, err := conn.client.Queues.Messages.Push(
+		context.TODO(),
+		conn.ep.CFQueue.QueueID,
+		queues.MessagePushParams{
+			AccountID: cloudflare.String(conn.ep.CFQueue.AccountID),
+			Body: queues.MessagePushParamsBodyMqQueueMessageText{
+				Body:        cloudflare.String(msg),
+				ContentType: cloudflare.F(queues.MessagePushParamsBodyMqQueueMessageTextContentTypeText),
+			},
+		},
+	)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func newCFQueueConn(ep Endpoint) *CFQueueConn {
+	return &CFQueueConn{
+		ep: ep,
+		t:  time.Now(),
+	}
+}

internal/endpoint/endpoint.go

@@ -42,6 +42,8 @@ const (
 	NATS = Protocol("nats")
 	// EventHub protocol
 	EventHub = Protocol("sb")
+	// CFQueue protocol
+	CFQueue = Protocol("cf-queue")
 )
 
 // Endpoint represents an endpoint.
@@ -134,6 +136,11 @@ type Endpoint struct {
 	EventHub struct {
 		ConnectionString string
 	}
+	CFQueue struct {
+		AccountID string
+		QueueID   string
+		APIToken  string
+	}
 	Local struct {
 		Channel string
 	}
@@ -241,6 +248,8 @@ func (epc *Manager) Send(endpoint, msg string) error {
 				conn = newLocalConn(ep, epc.publisher)
 			case EventHub:
 				conn = newEventHubConn(ep)
+			case CFQueue:
+				conn = newCFQueueConn(ep)
 			}
 			epc.conns[endpoint] = conn
 		}
@@ -298,6 +307,8 @@ func parseEndpoint(s string) (Endpoint, error) {
 		endpoint.Protocol = NATS
 	case strings.HasPrefix(s, "Endpoint="):
 		endpoint.Protocol = EventHub
+	case strings.HasPrefix(s, "cf-queue:"):
+		endpoint.Protocol = CFQueue
 	}
 
 	s = s[strings.Index(s, ":")+1:]
@@ -817,6 +828,48 @@ func parseEndpoint(s string) (Endpoint, error) {
 		endpoint.EventHub.ConnectionString = endpoint.Original
 	}
 
+	// Basic CF Queue connection strings in HOOKS interface
+	// cf-queue://<account_id>/<queue_id>?token=<api_token>
+	//
+	//  params are:
+	//
+	// token - API token
+	if endpoint.Protocol == CFQueue {
+		// Parse account_id/queue_id from the path parts
+		if len(sp) < 2 {
+			return endpoint, errors.New("invalid CF Queue format, should be account_id/queue_id")
+		}
+		endpoint.CFQueue.AccountID = sp[0]
+		endpoint.CFQueue.QueueID = sp[1]
+
+		// Parse query parameters for API token
+		if len(sqp) > 1 {
+			m, err := url.ParseQuery(sqp[1])
+			if err != nil {
+				return endpoint, errors.New("invalid CF Queue url")
+			}
+			for key, val := range m {
+				if len(val) == 0 {
+					continue
+				}
+				switch key {
+				case "token":
+					endpoint.CFQueue.APIToken = val[0]
+				}
+			}
+		}
+
+		if endpoint.CFQueue.AccountID == "" {
+			return endpoint, errors.New("missing CF Queue account ID")
+		}
+		if endpoint.CFQueue.QueueID == "" {
+			return endpoint, errors.New("missing CF Queue queue ID")
+		}
+		if endpoint.CFQueue.APIToken == "" {
+			return endpoint, errors.New("missing CF Queue API token")
+		}
+	}
+
 	return endpoint, nil
 }