feat(rx): add RxAssert with SendItems (#123)

.github/workflows/ci-workflow.yml

@@ -7,27 +7,27 @@ jobs:
     name: lint
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/setup-go@v3
+      - uses: actions/setup-go@v5
         with:
-          go-version: 1.21
+          go-version: 1.22
       - uses: actions/checkout@v3
       - name: golangci-lint
-        uses: golangci/golangci-lint-action@v3
+        uses: golangci/golangci-lint-action@v4
         with:
-          version: v1.54.2
+          version: v1.56.2
           args: --verbose
 
   test:
     strategy:
       matrix:
-        go-version: [1.21]
+        go-version: [1.22]
         platform: [ubuntu-latest, macos-latest]
 
     runs-on: ${{ matrix.platform }}
 
     steps:
       - name: Install Go
-        uses: actions/setup-go@v3
+        uses: actions/setup-go@v5
         with:
           go-version: ${{ matrix.go-version }}
 

.vscode/settings.json

@@ -6,12 +6,14 @@
   "cSpell.words": [
     "Assistable",
     "bodyclose",
+    "cmds",
     "coverpkg",
     "coverprofile",
     "cubiest",
     "deadcode",
     "depguard",
     "dogsled",
+    "dotenv",
     "dupl",
     "errcheck",
     "exportloopref",
@@ -27,6 +29,7 @@
     "goimports",
     "golangci",
     "goleak",
+    "Gomega",
     "gomnd",
     "goreleaser",
     "gosec",
@@ -43,11 +46,15 @@
     "nakedret",
     "nolint",
     "nolintlint",
+    "onecontext",
     "onsi",
+    "outdir",
     "passthru",
     "pixa",
     "prealloc",
     "repotoken",
+    "rxgo",
+    "samber",
     "shogo",
     "sidewalk",
     "skeletor",
@@ -56,12 +63,16 @@
     "staticcheck",
     "structcheck",
     "stylecheck",
+    "teivah",
+    "testcache",
     "thelper",
     "tparallel",
     "typecheck",
     "unconvert",
+    "Unmarshaller",
     "unparam",
     "varcheck",
+    "watchv",
     "watchvc",
     "watchvi",
     "wgan",

Taskfile.yml

@@ -18,6 +18,19 @@ tasks:
     cmds:
       - go build ./...
 
+  # NB: this clean task is very aggressive. It will not only remove
+  # old versions of packages, it removes all of them, even the
+  # current versions in use. You can use the download task to
+  # refresh the pkg cache.
+  #
+  clean:
+    cmds:
+      - go clean
+
+  clean-t:
+    cmds:
+      - go clean -testcache
+
   # === test =================================================
 
   t:
@@ -28,6 +41,14 @@ tasks:
     cmds:
       - go test ./i18n
 
+  tx:
+    cmds:
+      - ginkgo rx/...
+
+  tc:
+    cmds:
+      - ginkgo common/...
+
   dry:
     cmds:
       - ginkgo -v --dry-run ./...

boost/annotated-wait-group_test.go

@@ -3,7 +3,7 @@ package boost_test
 import (
 	"time"
 
-	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/ginkgo/v2" //nolint:revive // ginkgo ok
 
 	"github.com/snivilised/lorax/boost"
 )

boost/boost_suite_test.go

@@ -3,8 +3,8 @@ package boost_test
 import (
 	"testing"
 
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
+	. "github.com/onsi/ginkgo/v2" //nolint:revive // ginkgo ok
+	. "github.com/onsi/gomega"    //nolint:revive // gomega ok
 )
 
 func TestAsync(t *testing.T) {

boost/worker-pool.go

@@ -200,7 +200,9 @@ func (p *WorkerPool[I, O]) run(
 				// closure to forwardChOut, otherwise we end up in a deadlock.
 				//
 				running = false
+
 				close(forwardChOut)
+
 				p.Logger.Debug("source jobs chan closed",
 					slog.String("source", "worker-pool.run"),
 				)

boost/worker-pool_test.go

@@ -7,8 +7,8 @@ import (
 	"time"
 
 	"github.com/fortytw2/leaktest"
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
+	. "github.com/onsi/ginkgo/v2" //nolint:revive // ginkgo ok
+	. "github.com/onsi/gomega"    //nolint:revive // gomega ok
 	"github.com/samber/lo"
 
 	"github.com/snivilised/lorax/boost"
@@ -135,7 +135,7 @@ func (p *pipeline[I, O]) produce(parentContext context.Context,
 	interval time.Duration,
 	provider helpers.ProviderFunc[I],
 ) {
-	p.cancel = func(parentContext context.Context,
+	p.cancel = func(_ context.Context,
 		parentCancel context.CancelFunc,
 		delay time.Duration,
 	) {
@@ -144,8 +144,8 @@ func (p *pipeline[I, O]) produce(parentContext context.Context,
 			parentCancel,
 		)
 	}
-	p.stop = func(parentContext context.Context,
-		parentCancel context.CancelFunc,
+	p.stop = func(_ context.Context,
+		_ context.CancelFunc,
 		delay time.Duration,
 	) {
 		go helpers.StopProducerAfter(
@@ -235,15 +235,15 @@ var (
 		Eventually(parentContext, pipe.outputsDup.Channel).WithTimeout(time.Second * 5).Should(BeClosed())
 		Eventually(parentContext, pipe.producer.JobsCh).WithTimeout(time.Second * 5).Should(BeClosed())
 	}
-	assertCancelled assertFunc = func(parentContext context.Context,
-		pipe TestPipeline,
+	assertCancelled assertFunc = func(_ context.Context,
+		_ TestPipeline,
 		result *boost.PoolResult,
 	) {
 		// This needs to be upgraded to check that it is "timeout on send" error
 		Expect(result.Error).Error().NotTo(BeNil())
 	}
-	assertOutputChTimeout assertFunc = func(parentContext context.Context,
-		pipe TestPipeline,
+	assertOutputChTimeout assertFunc = func(_ context.Context,
+		_ TestPipeline,
 		result *boost.PoolResult,
 	) {
 		Expect(result.Error).Error().NotTo(BeNil())
@@ -257,14 +257,17 @@ var (
 		// }
 	}
 
-	assertNoError assertFunc = func(parentContext context.Context, pipe TestPipeline, result *boost.PoolResult) {
+	assertNoError assertFunc = func(_ context.Context,
+		_ TestPipeline, result *boost.PoolResult) {
 		Expect(result.Error).Error().To(BeNil())
 	}
 
 	assertAbort assertFunc = func(parentContext context.Context,
 		pipe TestPipeline,
 		result *boost.PoolResult,
 	) {
+		_, _ = parentContext, pipe
+
 		if result.Error == nil {
 			return // This is temporary
 		}

go.mod

@@ -29,6 +29,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/nicksnyder/go-i18n/v2 v2.3.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	github.com/teivah/onecontext v1.3.0
 	golang.org/x/exp v0.0.0-20230321023759-10a507213a29 // indirect
 	golang.org/x/net v0.20.0 // indirect
 	golang.org/x/sys v0.16.0 // indirect

go.sum

@@ -20,8 +20,11 @@ github.com/google/pprof v0.0.0-20230406165453-00490a63f317 h1:hFhpt7CTmR3DX+b4R1
 github.com/google/pprof v0.0.0-20230406165453-00490a63f317/go.mod h1:79YE0hCXdHag9sBkw2o+N/YnZtTkXi0UT9Nnixa5eYk=
 github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 h1:RWengNIwukTxcDr9M+97sNutRR1RKhG96O6jWumTTnw=
@@ -45,9 +48,13 @@ github.com/samber/lo v1.39.0/go.mod h1:+m/ZKRl6ClXCE2Lgf3MsQlWfh4bn1bz6CXEOxnEXn
 github.com/snivilised/extendio v0.6.1 h1:8IIJ4rryKcRv796QxfHcMnuWl4erFMqJF4Rqcj8eY/k=
 github.com/snivilised/extendio v0.6.1/go.mod h1:LbMvItdNqN8oZI4yUENxM459OkTP+W+hYXxKrc9l8ZQ=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/teivah/onecontext v1.3.0 h1:tbikMhAlo6VhAuEGCvhc8HlTnpX4xTNPTOseWuhO1J0=
+github.com/teivah/onecontext v1.3.0/go.mod h1:hoW1nmdPVK/0jrvGtcx8sCKYs2PiS4z0zzfdeuEVyb0=
+go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
@@ -56,21 +63,32 @@ go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
 go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
 go.uber.org/zap/exp v0.2.0 h1:FtGenNNeCATRB3CmB/yEUnjEFeJWpB/pMcy7e2bKPYs=
 go.uber.org/zap/exp v0.2.0/go.mod h1:t0gqAIdh1MfKv9EwN/dLwfZnJxe9ITAZN78HEWPFWDQ=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/exp v0.0.0-20230321023759-10a507213a29 h1:ooxPy7fPvB4kwsA2h+iBNHkAbp/4JxTSwCmvdjEYmug=
 golang.org/x/exp v0.0.0-20230321023759-10a507213a29/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
+golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo=
 golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
 golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
+golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.17.0 h1:FvmRgNOcs3kOa+T20R1uhfP9F6HgG2mfxDv1vrx1Htc=
 golang.org/x/tools v0.17.0/go.mod h1:xsh6VxdV005rRVaS6SSAf9oiAqljS7UZUacMZ8Bnsps=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscLw=
 google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

i18n/i18n_suite_test.go

@@ -3,8 +3,8 @@ package i18n_test
 import (
 	"testing"
 
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
+	. "github.com/onsi/ginkgo/v2" //nolint:revive // ginkgo ok
+	. "github.com/onsi/gomega"    //nolint:revive // gomega ok
 )
 
 func TestI18n(t *testing.T) {

rx/factory.go

@@ -0,0 +1,29 @@
+package rx
+
+// Amb takes several Observables, emit all of the items from only the first of these Observables
+// to emit an item or notification.
+func Amb[I any](observables []Observable[I], opts ...Option[I]) Observable[I] {
+	_, _ = observables, opts
+
+	panic("Amb: NOT-IMPL")
+}
+
+// FromChannel creates a cold observable from a channel.
+func FromChannel[I any](next <-chan Item[I], opts ...Option[I]) Observable[I] {
+	option := parseOptions(opts...)
+	ctx := option.buildContext(emptyContext)
+
+	return &ObservableImpl[I]{
+		parent:   ctx,
+		iterable: newChannelIterable(next, opts...),
+	}
+}
+
+func parseOptions[I any](opts ...Option[I]) Option[I] {
+	o := new(funcOption[I])
+	for _, opt := range opts {
+		opt.apply(o)
+	}
+
+	return o
+}