refactor outbound sender to allow for multiple modes of dispatch

caduceus.yaml

@@ -421,6 +421,21 @@ sender:
   # Defaults to 'false'.
   disablePartnerIDs: false
 
+# subcategory of senders that are predefined, listen for events and send to streams
+# instead of webhook callbacks
+outbound_stream_senders:
+   - PartnerIDs:
+     - comcast
+     - sky-italia
+     - cox
+     - rogers
+     Webhook:
+       Address: http://some-address
+       Events:
+         - ".*"
+
+
+
 # (Deprecated)
 # profilerFrequency: 15
 # profilerDuration: 15

dispatcher.go

@@ -0,0 +1,15 @@
+// SPDX-FileCopyrightText: 2021 Comcast Cable Communications Management, LLC
+// SPDX-License-Identifier: Apache-2.0
+
+package main
+
+import (
+	"container/ring"
+
+	"github.com/xmidt-org/wrp-go/v3"
+)
+
+type Dispatcher interface {
+	QueueOverflow()
+	Send(urls *ring.Ring, secret, acceptType string, msg *wrp.Message)
+}

go.mod

@@ -4,6 +4,7 @@ go 1.24
 
 require (
 	emperror.dev/emperror v0.33.0
+	github.com/aws/aws-sdk-go-v2/credentials v1.18.10
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc
 	github.com/go-kit/kit v0.13.0
 	github.com/gorilla/mux v1.8.1
@@ -28,11 +29,28 @@ require (
 	go.uber.org/zap v1.27.0
 )
 
+require (
+	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.1 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.6 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.6 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.6 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.6 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.29.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.34.2 // indirect
+	github.com/aws/smithy-go v1.23.0 // indirect
+)
+
 require (
 	emperror.dev/errors v0.8.1 // indirect
 	github.com/SermoDigital/jose v0.9.2-0.20161205224733-f6df55f235c2 // indirect
 	github.com/VividCortex/gohistogram v1.0.0 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.38.3
+	github.com/aws/aws-sdk-go-v2/config v1.31.6
+	github.com/aws/aws-sdk-go-v2/service/kinesis v1.40.1
+	github.com/aws/aws-sdk-go-v2/service/sts v1.38.2
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/billhathaway/consistentHash v0.0.0-20140718022140-addea16d2229 // indirect
 	github.com/c9s/goprocinfo v0.0.0-20210130143923-c95fcf8c64a8 // indirect
@@ -97,6 +115,7 @@ require (
 	github.com/xmidt-org/argus v0.10.18 // indirect
 	github.com/xmidt-org/arrange v0.4.0 // indirect
 	github.com/xmidt-org/chronon v0.1.4 // indirect
+	github.com/xmidt-org/retry v0.0.4
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/otel v1.36.0 // indirect
 	go.opentelemetry.io/otel/exporters/jaeger v1.17.0 // indirect

go.sum

@@ -679,8 +679,38 @@ github.com/aws/aws-sdk-go v1.55.7 h1:UJrkFq7es5CShfBwlWAC8DA077vp8PyVbQd3lqLiztE
 github.com/aws/aws-sdk-go v1.55.7/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU=
 github.com/aws/aws-sdk-go-v2 v0.18.0/go.mod h1:JWVYvqSMppoMJC0x5wdwiImzgXTI9FuZwxzkQq9wy+g=
 github.com/aws/aws-sdk-go-v2 v1.9.1/go.mod h1:cK/D0BBs0b/oWPIcX/Z/obahJK1TT7IPVjy53i/mX/4=
+github.com/aws/aws-sdk-go-v2 v1.38.3 h1:B6cV4oxnMs45fql4yRH+/Po/YU+597zgWqvDpYMturk=
+github.com/aws/aws-sdk-go-v2 v1.38.3/go.mod h1:sDioUELIUO9Znk23YVmIk86/9DOpkbyyVb1i/gUNFXY=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.1 h1:i8p8P4diljCr60PpJp6qZXNlgX4m2yQFpYk+9ZT+J4E=
+github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.1/go.mod h1:ddqbooRZYNoJ2dsTwOty16rM+/Aqmk/GOXrK8cg7V00=
+github.com/aws/aws-sdk-go-v2/config v1.31.6 h1:a1t8fXY4GT4xjyJExz4knbuoxSCacB5hT/WgtfPyLjo=
+github.com/aws/aws-sdk-go-v2/config v1.31.6/go.mod h1:5ByscNi7R+ztvOGzeUaIu49vkMk2soq5NaH5PYe33MQ=
+github.com/aws/aws-sdk-go-v2/credentials v1.18.10 h1:xdJnXCouCx8Y0NncgoptztUocIYLKeQxrCgN6x9sdhg=
+github.com/aws/aws-sdk-go-v2/credentials v1.18.10/go.mod h1:7tQk08ntj914F/5i9jC4+2HQTAuJirq7m1vZVIhEkWs=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.6 h1:wbjnrrMnKew78/juW7I2BtKQwa1qlf6EjQgS69uYY14=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.6/go.mod h1:AtiqqNrDioJXuUgz3+3T0mBWN7Hro2n9wll2zRUc0ww=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.6 h1:uF68eJA6+S9iVr9WgX1NaRGyQ/6MdIyc4JNUo6TN1FA=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.6/go.mod h1:qlPeVZCGPiobx8wb1ft0GHT5l+dc6ldnwInDFaMvC7Y=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.6 h1:pa1DEC6JoI0zduhZePp3zmhWvk/xxm4NB8Hy/Tlsgos=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.6/go.mod h1:gxEjPebnhWGJoaDdtDkA0JX46VRg1wcTHYe63OfX5pE=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3 h1:bIqFDwgGXXN1Kpp99pDOdKMTTb5d2KyU5X/BZxjOkRo=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.8.3/go.mod h1:H5O/EsxDWyU+LP/V8i5sm8cxoZgc2fdNR9bxlOFrQTo=
 github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.8.1/go.mod h1:CM+19rL1+4dFWnOQKwDc7H1KwXTz+h61oUSHyhV0b3o=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.1 h1:oegbebPEMA/1Jny7kvwejowCaHz1FWZAQ94WXFNCyTM=
+github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.1/go.mod h1:kemo5Myr9ac0U9JfSjMo9yHLtw+pECEHsFtJ9tqCEI8=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.6 h1:LHS1YAIJXJ4K9zS+1d/xa9JAA9sL2QyXIQCQFQW/X08=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.6/go.mod h1:c9PCiTEuh0wQID5/KqA32J+HAgZxN9tOGXKCiYJjTZI=
+github.com/aws/aws-sdk-go-v2/service/kinesis v1.40.1 h1:9QC0AF6gakV1TZuGp3NEUNl/6gXt3rfIifnxd+dWwbw=
+github.com/aws/aws-sdk-go-v2/service/kinesis v1.40.1/go.mod h1:UpSQbmXxFiDGDrvqsTgEm3YijDf9cg/Ti+s2W0SeFEU=
+github.com/aws/aws-sdk-go-v2/service/sso v1.29.1 h1:8OLZnVJPvjnrxEwHFg9hVUof/P4sibH+Ea4KKuqAGSg=
+github.com/aws/aws-sdk-go-v2/service/sso v1.29.1/go.mod h1:27M3BpVi0C02UiQh1w9nsBEit6pLhlaH3NHna6WUbDE=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.34.2 h1:gKWSTnqudpo8dAxqBqZnDoDWCiEh/40FziUjr/mo6uA=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.34.2/go.mod h1:x7+rkNmRoEN1U13A6JE2fXne9EWyJy54o3n6d4mGaXQ=
+github.com/aws/aws-sdk-go-v2/service/sts v1.38.2 h1:YZPjhyaGzhDQEvsffDEcpycq49nl7fiGcfJTIo8BszI=
+github.com/aws/aws-sdk-go-v2/service/sts v1.38.2/go.mod h1:2dIN8qhQfv37BdUYGgEC8Q3tteM3zFxTI1MLO2O3J3c=
 github.com/aws/smithy-go v1.8.0/go.mod h1:SObp3lf9smib00L/v3U2eAKG8FyQ7iLrJnQiAmR5n+E=
+github.com/aws/smithy-go v1.23.0 h1:8n6I3gXzWJB2DxBDnfxgBaSX6oe0d/t10qGz7OKqMCE=
+github.com/aws/smithy-go v1.23.0/go.mod h1:t1ufH5HMublsJYulve2RKmHDC15xu1f26kHCp/HgceI=
 github.com/benbjohnson/clock v1.0.3/go.mod h1:bGMdMPoPVvcYyt1gHDf4J2KE153Yf9BuiUKYMaxlTDM=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
@@ -1763,6 +1793,8 @@ github.com/xmidt-org/httpaux v0.3.2/go.mod h1:qmlPisXf80FTi3y4gX43eYbCVruSQyvu+F
 github.com/xmidt-org/httpaux v0.4.0/go.mod h1:UypqZwuZV1nn8D6+K1JDb+im9IZrLNg/2oO/Bgiybxc=
 github.com/xmidt-org/httpaux v0.4.2 h1:O6KTPy9Dtx6m5Ezb0nCQeAZWvBDgZvx82P16FSNauIE=
 github.com/xmidt-org/httpaux v0.4.2/go.mod h1:tZJ+SBoGNCxDOLopuSqrxaCkIVAQ+aPjNRf2XfMVwJA=
+github.com/xmidt-org/retry v0.0.4 h1:GUnMqjNUm2W0qfjmXS99keQ4qvyFsjh+HC66NXqLQfU=
+github.com/xmidt-org/retry v0.0.4/go.mod h1:Btl7o0Ts6iNEkF2liNiQepkkpHrK4rdGmqDVQ6KRHxo=
 github.com/xmidt-org/sallust v0.1.5/go.mod h1:azcKBypudADIeZ3Em8zGjVq3yQ7n4ueSvM/degHMIxo=
 github.com/xmidt-org/sallust v0.2.0/go.mod h1:HCQQn7po8czynjxtNVyZ5vzWuTqJyJwnPWkQoqBX67s=
 github.com/xmidt-org/sallust v0.2.1/go.mod h1:68C0DLwD5xlhRznXTWmfUhx0etyrFpSOzYGU7jzmpzs=

internal/batch/batch.go

@@ -0,0 +1,40 @@
+// SPDX-FileCopyrightText: 2023 Comcast Cable Communications Management, LLC
+// SPDX-License-Identifier: Apache-2.0
+
+package batch
+
+import ()
+
+func GetBatch[T any](start int, batchSize int, items []T) ([]T, bool) {
+	var batch []T
+	more := true
+	if (len(items) - start) <= batchSize {
+		batch = items[start:]
+		more = false
+	} else {
+		batch = items[start : start+batchSize]
+	}
+
+	return batch, more
+}
+
+func GetBatches[T any](batchSize int, items []T) [][]T {
+	batches := [][]T{}
+
+	if len(items) == 0 {
+		return batches
+	}
+
+	notDone := true
+	start := 0
+	for notDone {
+		items, more := GetBatch(start, batchSize, items)
+		notDone = more
+		start += batchSize
+
+		batches = append(batches, items)
+	}
+
+	return batches
+
+}

internal/batch/batch_submitter.go

@@ -0,0 +1,8 @@
+// SPDX-FileCopyrightText: 2023 Comcast Cable Communications Management, LLC
+// SPDX-License-Identifier: Apache-2.0
+
+package batch
+
+type BatchSubmitter[T any] interface {
+	SubmitBatch(events []T) error
+}

internal/batch/batch_test.go

@@ -0,0 +1,115 @@
+// SPDX-FileCopyrightText: 2023 Comcast Cable Communications Management, LLC
+// SPDX-License-Identifier: Apache-2.0
+
+package batch
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGetBatch(t *testing.T) {
+	batchSize := 5
+
+	items := []string{}
+	batchOfItems, more := GetBatch(0, batchSize, items)
+	assert.False(t, more)
+	assert.Equal(t, 0, len(batchOfItems))
+
+	items = []string{"a", "b", "c", "d"}
+	batchOfItems, more = GetBatch(0, batchSize, items)
+	assert.False(t, more)
+	assert.Equal(t, 4, len(batchOfItems))
+
+	items = []string{"a", "b", "c", "d", "e"}
+	batchOfItems, more = GetBatch(0, batchSize, items)
+	assert.False(t, more)
+	assert.Equal(t, 5, len(batchOfItems))
+
+	items = []string{"a", "b", "c", "d", "e", "f"}
+	batchOfItems, more = GetBatch(0, batchSize, items)
+	assert.True(t, more)
+	assert.Equal(t, 5, len(batchOfItems))
+	batchOfItems, more = GetBatch(0+batchSize, batchSize, items)
+	assert.False(t, more)
+	assert.Equal(t, 1, len(batchOfItems))
+}
+
+func TestGetLargeBatch(t *testing.T) {
+	batchSize := 5
+
+	start := 0
+	items := []string{"a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k"}
+	batchOfItems, more := GetBatch(start, batchSize, items)
+	assert.True(t, more)
+	assert.Equal(t, 5, len(batchOfItems))
+
+	start += batchSize
+	batchOfItems, more = GetBatch(start, batchSize, items)
+	assert.True(t, more)
+	assert.Equal(t, 5, len(batchOfItems))
+
+	start += batchSize
+	batchOfItems, more = GetBatch(start, batchSize, items)
+	assert.False(t, more)
+	assert.Equal(t, 1, len(batchOfItems))
+}
+
+func TestGetBatches100Items(t *testing.T) {
+	batchSize := 25
+
+	items := []string{}
+
+	for i := 0; i < 100; i++ {
+		items = append(items, fmt.Sprintf("%d", i))
+	}
+
+	batches := GetBatches(batchSize, items)
+	assert.Equal(t, 4, len(batches))
+	assert.Equal(t, 25, len(batches[0]))
+	assert.Equal(t, 25, len(batches[1]))
+	assert.Equal(t, 25, len(batches[2]))
+	assert.Equal(t, 25, len(batches[3]))
+}
+
+func TestGetBatchesEmptyItems(t *testing.T) {
+	batchSize := 25
+
+	items := []string{}
+
+	batches := GetBatches(batchSize, items)
+	assert.Equal(t, 0, len(batches))
+}
+
+func TestGetBatchesLessThanOne(t *testing.T) {
+	batchSize := 5
+
+	items := []string{"a", "b", "c", "d"}
+
+	batches := GetBatches(batchSize, items)
+	assert.Equal(t, 1, len(batches))
+	assert.Equal(t, 4, len(batches[0]))
+}
+
+func TestGetBatchesExactlyOne(t *testing.T) {
+	batchSize := 5
+
+	items := []string{"a", "b", "c", "d", "e"}
+
+	batches := GetBatches(batchSize, items)
+	assert.Equal(t, 1, len(batches))
+	assert.Equal(t, 5, len(batches[0]))
+}
+
+func TestGetBatchesOneOver(t *testing.T) {
+	batchSize := 5
+
+	items := []string{"a", "b", "c", "d", "e", "f"}
+
+	batches := GetBatches(batchSize, items)
+	assert.Equal(t, 2, len(batches))
+	assert.Equal(t, 5, len(batches[0]))
+	assert.Equal(t, 1, len(batches[1]))
+}