test(rx): add has items matcher (#266)

Author: plastikfan

.vscode/settings.json

@@ -24,6 +24,7 @@
     "godirwalk",
     "gofmt",
     "goimports",
+    "goleak",
     "gomnd",
     "gosec",
     "gosimple",
@@ -43,14 +44,17 @@
     "navi",
     "nicksnyder",
     "nolintlint",
+    "onecontext",
     "prealloc",
     "repath",
     "Resumer",
+    "rxgo",
     "sidewalk",
     "staticcheck",
     "structcheck",
     "stylecheck",
     "Syncer",
+    "teivah",
     "thelper",
     "tparallel",
     "typecheck",

README.md

@@ -359,6 +359,14 @@ The ___Start___ and ___Stop___ conditions are defined using `ListenBy`, eg:
 
 ## 🔨 Development
 
+___RxGo___
+
+<p align="left">
+  <a href="https://rxjs.dev/guide/overview"><img src="https://avatars.githubusercontent.com/u/6407041?s=200&v=4" width="50" /></a>
+</p>
+
+To support concurrency features, Extendio uses the reactive model provided by [RxGo](https://github.com/ReactiveX/RxGo). However, since ___RxGo___ seems to be a dead project with its last release in April 2021 and its unit tests not currently running successfully, the decision has been made to re-implement this locally. One of the main reasons for the project no longer being actively maintained is the release of generics feature in Go version 1.18, and supporting generics in RxGo would require significant effort to re-write the entire library. While work on this has begun, it's unclear when this will be delivered. Despite this, the reactive model's support for concurrency is highly valued, and Extendio aims to make use of a minimal functionality set for parallel processing during directory traversal. The goal is to replace it with the next version of RxGo when it becomes available.
+
 See:
 
 - [Github Development Workflow](./resources/doc/GITHUB-DEV.md)

go.mod

@@ -14,6 +14,7 @@ require (
 require (
 	github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect
 	github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38 // indirect
+	go.uber.org/goleak v1.2.1 // indirect
 	golang.org/x/tools v0.8.0 // indirect
 )
 
@@ -23,6 +24,7 @@ require (
 	github.com/natefinch/lumberjack v2.0.0+incompatible
 	github.com/nicksnyder/go-i18n/v2 v2.2.1
 	github.com/pkg/errors v0.9.1
+	github.com/teivah/onecontext v1.3.0
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
 	golang.org/x/exp v0.0.0-20220303212507-bbda1eaf7a17 // indirect

go.sum

@@ -17,6 +17,9 @@ github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeN
 github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38 h1:yAJXTCF9TqKcTiHJAE8dj7HMvPfh66eeA2JYW7eFpSE=
 github.com/google/pprof v0.0.0-20210407192527-94a9f03dee38/go.mod h1:kpwsk12EmLew5upagYY7GY0pfYCcupk39gWOCRROcvE=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+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/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 h1:RWengNIwukTxcDr9M+97sNutRR1RKhG96O6jWumTTnw=
 github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826/go.mod h1:TaXosZuwdSHYgviHp1DAtfrULt5eUgsSMsZf+YrPgl8=
 github.com/natefinch/lumberjack v2.0.0+incompatible h1:4QJd3OLAMgj7ph+yZTuX13Ld4UpgHp07nNdFX7mqFfM=
@@ -34,12 +37,18 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/samber/lo v1.38.1 h1:j2XEAqXKb09Am4ebOg31SpvzUTTs6EN3VfgeLUhPdXM=
 github.com/samber/lo v1.38.1/go.mod h1:+m/ZKRl6ClXCE2Lgf3MsQlWfh4bn1bz6CXEOxnEXnEA=
 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.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/teivah/onecontext v1.3.0 h1:tbikMhAlo6VhAuEGCvhc8HlTnpX4xTNPTOseWuhO1J0=
+github.com/teivah/onecontext v1.3.0/go.mod h1:hoW1nmdPVK/0jrvGtcx8sCKYs2PiS4z0zzfdeuEVyb0=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
 go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
+go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
 go.uber.org/goleak v1.1.11 h1:wy28qYRKZgnJTxGxvye5/wgWr1EKjmUDGYox5mGlRlI=
+go.uber.org/goleak v1.2.1 h1:NBol2c7O1ZokfZ0LEU9K6Whx/KnwvepVetCUhtKja4A=
+go.uber.org/goleak v1.2.1/go.mod h1:qlT2yGI9QafXHhZZLxlSuNsMw3FFLxBr+tBRlmO1xH4=
 go.uber.org/multierr v1.9.0 h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI=
 go.uber.org/multierr v1.9.0/go.mod h1:X2jQV1h+kxSjClGpnseKVIxpmcjrj7MNnI0bnlfKTVQ=
 go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60=
@@ -48,7 +57,9 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/exp v0.0.0-20220303212507-bbda1eaf7a17 h1:3MTrJm4PyNL9NBqvYDSj3DHl46qQakyfqfWo4jgfaEM=
 golang.org/x/exp v0.0.0-20220303212507-bbda1eaf7a17/go.mod h1:lgLbSvA5ygNOMpwM/9anMpWVlVJ7Z+cHWq/eFuinpGE=
+golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
+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.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
@@ -73,6 +84,8 @@ golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+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.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.8.0 h1:vSDcovVPld282ceKgDimkRSC8kpaH1dgyc9UMzlt84Y=
@@ -81,8 +94,10 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscLw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc=
 gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
 gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

rx/item.go

@@ -0,0 +1 @@
+This implementation is based upon the original version of rxgo at https://github.com/ReactiveX/RxGo, based upon generics rather than relying on reflection and interface{}. It is intended to replace this local version with the official version when and if it becomes available and for this to happen with minimal issues, the design here has to mirror the original as closely as possible.

rx/item_test.go

@@ -0,0 +1,12 @@
+package rxgo
+
+// FromChannel creates a cold observable from a channel.
+func FromChannel[T any](next <-chan Item[T], opts ...Option[T]) Observable[T] {
+	option := parseOptions(opts...)
+	ctx := option.buildContext(emptyContext)
+
+	return &ObservableImpl[T]{
+		parent:   ctx,
+		iterable: newChannelIterable(next, opts...),
+	}
+}

rxgo/README.txt

@@ -0,0 +1,96 @@
+package rxgo
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/samber/lo"
+)
+
+// CloseChannelStrategy indicates a strategy on whether to close a channel.
+type CloseChannelStrategy uint32
+
+const (
+	// LeaveChannelOpen indicates to leave the channel open after completion.
+	LeaveChannelOpen CloseChannelStrategy = iota
+	// CloseChannel indicates to close the channel open after completion.
+	CloseChannel
+)
+
+type Item[T any] struct {
+	V T
+	E error
+}
+
+// Of creates an item from a value.
+func Of[T any](i T) Item[T] {
+	return Item[T]{V: i}
+}
+
+// Error checks if an item is an error.
+func (i Item[T]) Error() bool {
+	return i.E != nil
+}
+
+// SendItemsV is an utility function that send a list of items and indicate
+// a strategy on whether to close the channel once the function completes.
+func SendItems[T any](ctx context.Context, ch chan<- Item[T], strategy CloseChannelStrategy, items ...T) {
+	if strategy == CloseChannel {
+		defer close(ch)
+	}
+
+	send(ctx, ch, items...)
+}
+
+func send[T any](ctx context.Context, ch chan<- Item[T], items ...T) {
+	// This is only the basic implementation. It does not yet support sending
+	// a slice or a channel. Support for these don't need to be added unless
+	// explicitly required.
+	//
+	for _, currentItem := range items {
+		_ = Of(currentItem).SendContext(ctx, ch)
+	}
+}
+
+// SendItemsV (verbose) is an utility function that send a list of items and indicate
+// a strategy on whether to close the channel once the function completes.
+// This will eventually be removed as its only required for debugging purposes
+func SendItemsV[T any](ctx context.Context, ch chan<- Item[T], strategy CloseChannelStrategy, items ...T) {
+	if strategy == CloseChannel {
+		defer close(ch)
+	}
+
+	sendV(ctx, ch, items...)
+}
+
+// This will eventually be removed as its only required for debugging purposes
+func sendV[T any](ctx context.Context, ch chan<- Item[T], items ...T) {
+	fmt.Println("")
+
+	// This is only the basic implementation. It does not yet support sending
+	// a slice or a channel. Support for these don't need to be added unless
+	// explicitly required.
+	//
+	for _, currentItem := range items {
+		result := Of(currentItem).SendContext(ctx, ch)
+		indicator := lo.Ternary(result, "✔️", "❌")
+
+		fmt.Printf("===> 💘 sending item: '%v' (%v)\n", currentItem, indicator)
+	}
+}
+
+// SendContext sends an item and blocks until it is sent or a context canceled.
+// It returns a boolean to indicate whether the item was sent.
+func (i Item[T]) SendContext(ctx context.Context, ch chan<- Item[T]) bool {
+	select {
+	case <-ctx.Done(): // Context's done channel has the highest priority
+		return false
+	default:
+		select {
+		case <-ctx.Done():
+			return false
+		case ch <- i:
+			return true
+		}
+	}
+}

rxgo/factory.go

@@ -0,0 +1,52 @@
+package rxgo_test
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	. "github.com/onsi/gomega/matchers"
+	"github.com/onsi/gomega/types"
+
+	"github.com/snivilised/extendio/rxgo"
+)
+
+// TODO: check to see if we need to use async ginkgo types
+
+var _ = Describe("Item", func() {
+	Context("Variadic", func() {
+		Context("AndMatcher", func() {
+			It("🧪 should: Send Items", func() {
+				ch := make(chan rxgo.Item[int], 3)
+				actual := RxObservable[int]{
+					context:    context.Background(),
+					observable: rxgo.FromChannel(ch),
+				}
+
+				go rxgo.SendItemsV(context.Background(), ch, rxgo.CloseChannel, 1, 2, 3)
+				Expect(actual).Should(&AndMatcher{
+					Matchers: []types.GomegaMatcher{
+						MatchHasItems(1, 2, 3),
+					},
+				})
+			})
+		})
+
+		Context("Singular MatchHasItems", func() {
+			It("🧪 should: Send Items", func() {
+				ch := make(chan rxgo.Item[int], 3)
+				actual := RxObservable[int]{
+					context:    context.Background(),
+					observable: rxgo.FromChannel(ch),
+				}
+
+				go rxgo.SendItems(context.Background(), ch, rxgo.CloseChannel, 1, 2, 3)
+				Expect(actual).Should(MatchHasItems(1, 2, 3))
+			})
+		})
+	})
+
+	Context("Variadic with Error", func() {
+		It("🧪 should: (can't send an error into a channel)", Pending)
+	})
+})