Cold observable

[teivah]

A first implementation of cold observables.

First, I changed the few things:

• Iterator remains an interface but proposes two methods Next() bool and Value() interface{}. No real difference here but just a more idiomatic way to iterate on a structure, in my opinion.
• Iterable became an interface with only one Iterator() Iterator method.
• Observable became an interface with a list of operators but is also composed of Iterable.

To iterate over an Observable, every operator is doing it this way:

it := observable.iterable.Iterator()
for it.Next() {
  item := it.Value()
  // Do something
}

Then, I tried to respect two approaches:

• The initial idea of cold observables described here (Observables are functions that tie an observer to a producer).
• The initial implementation of @jochasinga based on Golang pipelines.

For example, let's check at the Map operator implementation:

f := func(out chan interface{}) {
	it := o.Iterator()
	for it.Next() {
		item := it.Value()
		out <- apply(item)
	}
	close(out)
}

return newColdObservable(f)

Instead of creating a goroutine once the operator is called, we wrap a function within the Observable returned.

When did this function trigger? Basically, during a subscription to the Observable. This allows to reproduce the lazy behavior of cold Observable and to keep the initial idea of managing pipelines between goroutines.

One unit test is worth thousand words:

just := Just(1).Map(func(i interface{}) interface{} {
	return 1 + i.(int)
}).Map(func(i interface{}) interface{} {
	return 1 + i.(int)
})

AssertThatObservable(t, just, HasItems(3))
AssertThatObservable(t, just, HasItems(3)) // Previously, the Observable was emitting nothing during the second subscription

@avelino Please review this one before #157

[coveralls]

Pull Request Test Coverage Report for Build 299

• 804 of 815 (98.65%) changed or added relevant lines in 7 files are covered.
• 29 unchanged lines in 2 files lost coverage.
• Overall coverage decreased (-0.6%) to 95.527%

---

Changes Missing Coverage	Covered Lines	Changed/Added Lines	%
observablecreate.go	84	86	97.67%
connectableobservable.go	6	9	66.67%
single.go	38	44	86.36%

Files with Coverage Reduction	New Missed Lines	%
observablecreate.go	1	94.48%
observable.go	28	97.24%

Totals
Change from base Build 267:	-0.6%
Covered Lines:	1730
Relevant Lines:	1811

---

💛 - Coveralls

[jochasinga]

Thanks @teivah for the effort!

Is there a reason for having separate Next() and Value() methods for Iterator? The use in the for loop is pretty neat (and is less verbose) but I think it's less idiomatic than something like this:

for {
        if v, ok := it.Next(); ok {
                 // Do something with v
        } else {
                fmt.Println("End of iterator")
                break
        }
}

The decision to implement Iterator.Next() (interface{}, error) was to keep it idiomatic (Go's multiple returns, as well as correlation to some other language's Iterator behavior like JS). I also like returning error better than boolean since it's less likely to be misused and provide some information.

[teivah]

@jochasinga Hi, nice to discuss with you :)

Actually, I did it because from my perspective it was even more idiomatic ^^

As a concrete example, taken from Go standard library: bufio.Scanner. It proposes the two following methods:

• Scan() bool
• Text() string

So, to iterate over a Scanner, we have to do something like this:

for myScanner.Scan() {
  value := myScanner.Text()
  fmt.Println(value)
}

But obviously this is not that important. If you prefer to keep it as you initially proposed, I can change it.

[jochasinga]

The Scanner interface makes sense since it deals with parsing tokens. However, Iterator shares more with generators and iterators.

In Python, a generator implements a next() method which yields the "next" value and raises an exception at the end. Multiple return is not Pythonic (it's actually tuple-ic) so next() only yield a value OR error. The key here is it only implements next() to get the value.

In JS however, an iterator next() method actually yield a result object like {value: 'someVal', done: false}. That means it is closer to what you're proposing in the sense that you have to access a value property to access the next value. However, I find this implementation a bit awkward because you can't atomically update value and done. What if there's the next value waiting in line but done turns out true?

Same goes with Value() and Next() because once the user actually uses an Iterator, Next() could returns false but Value() might still be able to yield a next value. This might be easily preventable at the implementation level but at the user level I feel that it's not worth the superficial verbosity decrease.

In addition, I think a method Next() that returns a bool is misleading. The semantic should be HasNext(), which would then violate the common Iterator API. 😅

IMHO I'm a fan of

if val, err := iter.Next(); err != nil {
        // ...
}

than returning a bool as the second value because like I said earlier an error carries over some information (in this case, an IteratorEnd error could be implemented) than the former.

cc @avelino

[teivah]

because like I said earlier an error carries over some information

Okay I see the potential benefit. I did a rollback to use the previous Iterator interface (only one Next() (interface{}, error) method)

[ghost]

Found some fixes!

P.S. share your ideas, feedbacks or issues with us at https://github.com/fixmie/feedback (this message will be removed after the beta stage).

[teivah]

@avelino Ready for 2019! 😄