Outer Product

[CJ-Wright]

Here is my attempt at an example:

from streams import Streams
from operator import mul
s1 = Stream()
s2 = Stream()

op = s1.product(s2, mul)
L = op.sink_to_list()
a = [1, 2, 3]
b = [4, 5, 6]

for x, y in zip(a, b):
    s1.emit(x)
    s2.emit(y)
print(L)

[4, 5, 6, 8, 10, 12, 12, 15, 18]

The best way that I can kind of come up with this is to stash all but one of the streams into a list and then for each incoming piece of data perform an itertools.product on the resulting iterable and push it into some function. I'm not so crazy about this as it means we end up storing a bunch of stuff in memory which seems un-stream like.

[mrocklin]

Can I suggest that this is an extreme case of a streaming join operation. In a streaming join operation you're going to hold on to some backlog of events in each stream and when an event occurs in the others you emit if some condition is met. Full outer product is this case when the condition is lambda *args: True and the backlog is infinite.

When thinking about streaming joins there are now questions about how to define the length of a backlog. Is it a number of elements? Is it something having to do with the data itself, like a time value?

[CJ-Wright]

Hmm ok I like that vision of the problem.
As a side note we may want to look to itertools to find new nodes to implement.

Keeping track of the number of elements is ok. The timed one would mean that we would need to track at what time the data came down and then cull the data as it became to old, which may be possible. We could even ask the user to provide a pair of functions, one which assigns a value to the data and the other which asses if the value invalidates the data.

egs

source = Stream()
source2 = Stream()
timeout_node = source.conditional_product(source2, assigning_function=time, deciding function=lambda x: (time() - x) < 60)

In this case if the data is older than a minute then we remove it.

source = Stream()
source2 = Stream()
CountClass():
    def __call__():
        self.i += 1
        return self.i
countout_node = source.conditional_product(source2, assigning_function=CountClass, deciding function=lambda x: x + 5 < CountClass.i)

source = Stream()
source2 = Stream()
outer_product = source.conditional_product(source2, assigning_function=lambda *args: True, deciding function=lambda x: x, emit_on=source)

[mrocklin]

As a side note we may want to look to itertools to find new nodes to implement.

Sure, I'm biased, but toolz may also have operations (like streaming joins). However, I'm also inclined to only add things as they become necessary. There is a cost to adding and maintaining functionality.

Similarly for time-based joins I would say that we shouldn't deal with it until we have a reason to. Number-of-element buffers are probably more sensible to deal with.

[CJ-Wright]

That's fair.