rx: fix Join operator

[plastikfan]

Join operator is a bit more complicated to port over, due to its use of int64 to represent a time value:

// Returns absolute value for int64
func abs(n int64) int64 {
	y := n >> 63
	return (n ^ y) - y
}

The join algorithm performs bit shifting on time values (window duration), which are sent through channel. So how do we shoehorn this into the new representation of Item[T]?

Introducing a new time value member on Item, causes rippling changes accross different types like Func. This problem needs to be fixed into its own issue.

[plastikfan]

This was the first attempt of implementing Join, that was abandoned:

// Returns absolute value for int
func absN(n int) int {
	sz := unsafe.Sizeof(int(0)) / 8
	y := n >> sz

	return (n ^ y) - y
}

// Join combines items emitted by two Observables whenever an item from one
// Observable is emitted during a time window defined according to an item
// emitted by the other Observable. The time is extracted using a
// timeExtractor function.
func (o *ObservableImpl[T]) Join(joiner Func2[T], right Observable[T],
	timeExtractor TimeExtractorFunc[T], window Duration, opts ...Option[T],
) Observable[T] {
	f := func(ctx context.Context, next chan Item[T], option Option[T], opts ...Option[T]) {
		defer close(next)
		windowDuration := window.duration()
		rBuf := make([]Item[T], 0)

		lObserve := o.Observe()
		rObserve := right.Observe()
	lLoop:
		for {
			select {
			case <-ctx.Done():
				return
			case lItem, ok := <-lObserve:
				if lItem.V == nil && !ok {
					return
				}

				if lItem.IsError() {
					lItem.SendContext(ctx, next)
					if option.getErrorStrategy() == StopOnError {
						return
					}
					continue
				}

				lTime := timeExtractor(lItem.V).UnixNano()
				cutPoint := 0

				for i, rItem := range rBuf {
					rTime := timeExtractor(rItem.V).UnixNano()

					if abs(lTime-rTime) <= windowDuration {
						i, err := joiner(ctx, lItem.V, rItem.V)

						if err != nil {
							Error[T](err).SendContext(ctx, next)

							if option.getErrorStrategy() == StopOnError {
								return
							}
							continue
						}
						Of(i).SendContext(ctx, next)
					}

					if lTime > rTime+windowDuration {
						cutPoint = i + 1
					}
				}

				rBuf = rBuf[cutPoint:]

				for {
					select {
					case <-ctx.Done():
						return
					case rItem, ok := <-rObserve:
						if rItem.V == nil && !ok {
							continue lLoop
						}
						if rItem.IsError() {
							rItem.SendContext(ctx, next)

							if option.getErrorStrategy() == StopOnError {
								return
							}
							continue
						}

						rBuf = append(rBuf, rItem)
						rTime := timeExtractor(rItem.V).UnixNano()

						if abs(lTime-rTime) <= windowDuration {
							i, err := joiner(ctx, lItem.V, rItem.V)
							if err != nil {
								Error[T](err).SendContext(ctx, next)
								if option.getErrorStrategy() == StopOnError {
									return
								}
								continue
							}
							Of(i).SendContext(ctx, next)

							continue
						}
						continue lLoop
					}
				}
			}
		}
	}

	return customObservableOperator(o.parent, f, opts...)
}