add rx

[plastikfan]

This is the overall issue covering adding support for RX (a stop-gap for rxgo-v3)

• add gomega RxAssert #123
• rx: add optional single #125
• resolve gather #131
• resolve SendItems #132
• complete factory implementation #134
• rx: add Amb #136
• rx: add factory connectable tests #139
• rx: revert Map from only working with number #141
• rx: rationalise Item auxilliary fields #142
• rx: improve assert design #144
• rx: finish factory connectable tests #147
• Generate operator test skeleton #149

The following is a list of the operators to be implemented

• rx: All operator #152
• rx: Average operator #154
• rx: BackOffRetry operator #156
• rx: BufferWithCount / BufferWithTime operator
• rx: Contains operator #158
• rx: Count operator #160
• rx: Debounce
• rx: DefaultIfEmpty operator #162
• rx: Distinct operator #164
• rx: DoOnCompleted / DoOnError / DoOnNext operator #166
• rx: ElementAt operator #168
• rx: Error / Errors operator #170
• rx: Filter operator #172
• rx: Find operator #174
• rx: First operator #176
• rx: FlatMap, ForEach, IgnoreElements, GroupBy, Last operators #178
• rx: Map, Max and Min operators #181
• rx: OnErrorResumeNext / OnErrorReturn / OnErrorReturnItem operator and Observe operators #184
• rx: Reduce operator #186
• rx: Repeat operator #189
• rx: Retry operator #192
• rx: Run operator #194
• rx: Sample operator #196
• rx: Scan operator #198
• rx: SequenceEqual operator #200
• rx: Send operator #202
• rx: Serialize operator #204
• rx: Skip / SkipLast / SkipWhile operator #206
• rx: StartWith operator #208
• rx: Sum operator #210
• rx: Take / TakeLast / TakeUntil / TakeWhile operator #212
• rx: Time operators #214
• rx: ToMap / ToMapWithValueSelector operator
• rx: ToSlice operator #216
• rx: UnMarshal operator
• rx: WindowWith operators #218
• rx: ZipFromIterable operator #220

Improvements

• rx: optimise item struct #221
• insert rxjs copyright notices #224
• add a try item assister object #227
• define new Bridge operator to complement Map #230

Fixes

• rx: fix Join operator #179
• rx: Marshal operator needs multi-type Mapping #182
• rx: define mocks #188
• fix parallel #226
• rx: allow client control of reduce accumulator #231
• rx: change Range operator to be able work with native instances of Item[T] #232

Also, see this issue on rxgo: Support for generics

[plastikfan]

In rxgo the test 'Test_Observable_Debounce' fails:

=== RUN   Test_Observable_Debounce
    /Users/plastikfan/dev/github/go/third/rxgo/assert.go:225:
                Error Trace:    /Users/plastikfan/dev/github/go/third/rxgo/assert.go:225
                                                        /Users/plastikfan/dev/github/go/third/rxgo/observable_operator_test.go:389
                Error:          Not equal:
                                expected: []interface {}{1, 2, 5, 6}
                                actual  : []interface {}{1, 2, 5}

                                Diff:
                                --- Expected
                                +++ Actual
                                @@ -1,6 +1,5 @@
                                -([]interface {}) (len=4) {
                                +([]interface {}) (len=3) {
                                  (int) 1,
                                  (int) 2,
                                - (int) 5,
                                - (int) 6
                                + (int) 5
                                 }
                Test:           Test_Observable_Debounce
--- FAIL: Test_Observable_Debounce (0.00s)
FAIL
FAIL    github.com/reactivex/rxgo/v2    1.243s

This will not be addressed unless it affects the requirements of snivilised projects.

[plastikfan]

rx.Assert needs to be modified to accommodate the new ways an item can be created: Ch, Tick, Tv. Probably need some kind of generic reader function that implements this to read and interpret an existing item that can be created by one of these alternative methods. Probably need to define new assertions, ie HasCh, HasTick, HasTv.

Perhaps in iterable.go, we define a base struct iterable-X that contains a function to read each item.

[plastikfan]

There is a lot of option copying code; should really define a function to do this, rather than have cloned code in multiple places; eg (i *createIterable[T]) Observe in iterable-create.go.

The current definition of Item:

	Item[T any] struct {
		V T
		E error
		//
		C       chan<- Item[T]
		tick    bool
		tickV   bool
		numeric bool
		TV      int
		N       int
	}

can easily be optimised by replacing all the boolean flags with a pseudo bitwsie enum (discriminator) and because N and TV are both int, they can be replaced with a common single field, actually, this could still be called N, and we use the discriminator to discover how N should be interpreted. If the discriminiator is zero, then we know the item is just a T. We only add fields to Item, if it is of a fixed type and not of any existing type (unless multiple fields are required for a single purpose).

[plastikfan]

observble.go contains the runParallel function which implements a worker pool. Perhaps this can be modified to use the lorax.WorkerPool.

Actually, the above statement is incorrect. The WorkerPool is function based, where as the pool in rx is event/data oriented, so the 2 are based on 2 different concepts. However, it would be nice if there was an explicit representation of the rx pool with a clear interface rather than it being buried inside runParallel (scatter).

[plastikfan]

We might be better off turning Item into an interface, then having multiple value types that implement this interface, rather than the current implementation which is a struct with multiple fields.

Actually, this might not be a good idea after all, since if we introduce an interface, then:

• the interface having multiple IsXXX methods, with each concrete type implementing true to 1 of these and false for all the others. introducing new types would mean rippling changes accross all those types. This design would not be scalable
• the interface having a getter method for the underlying value could not be implemented, because it could not be narrowed down into a common type other than any, which defeats the point of using generics

For these reasons, we'll keep the current implementation, accepting that there is an overhead of field values for every item.

[plastikfan]

There is a lot of option copying code; should really define a function to do this, rather than have cloned code in multiple places; eg *(i createIterable[T]) Observe in iterable-create.go.

The current definition of Item:

	Item[T any] struct {
		V T
		E error
		//
		C       chan<- Item[T]
		tick    bool
		tickV   bool
		numeric bool
		TV      int
		N       int
	}

can easily be optimised by replacing all the boolean flags with a pseudo bitwsie enum (discriminator) and because N and TV are both int, they can be replaced with a common single field, actually, this could still be called N, and we use the discriminator to discover how N should be interpreted. If the discriminiator is zero, then we know the item is just a T. We only add fields to Item, if it is of a fixed type and not of any existing type (unless multiple fields are required for a single purpose).

Actually, we could optimise this further. Item is now currently defined as:

	Item[T any] struct {
		V T
		E error
		//
		C    chan<- Item[T]
		N    NumVal
		B    bool
		O    any
		disc enums.ItemDiscriminator
	}

but with the intruction of Opaque/O, it becomes apparent, that there is now no need for the overhead of haing multiple fields for different purposes. All we need is a singlew Opqaue field along with the discriminator. All the other fields can be discarded, reducing the size of Item.

To support this approach, we could define all those Item creational functions as getters on Item, eg, Item[T].Of, would be the complement of the Of[T], in that it gets the value as a T from an exiting item.