Motivation

• Adds support for nats streaming

Modifications

• Addition of a new package, nats

Verify change

• [ ] Make sure the change passes the CI checks.

This is probably not high priority for the vast majority of users, however the use of the confluent kafka client makes it so that go-streams is not windows compatible out of the box.

The suggestion I'd like to make is to change the extension to use something like: https://github.com/segmentio/kafka-go

I wonder whether this library supports loop/feedback constructs. Skimming the API gave a hint it shall be possible but I wonder about real world implications.

Any insights? Thanks!

In all examples the main go-routine either blocks forever or waits for timeout.

Is there any way to gracefully exit main function once processing completes (but not aborting all currently running go-routines until they finish the job)? I can do it with ChanSink and using some sync primitive inside it, but it looks a bit ugly. Any nicer solution?

Motivation

• Fixes #___

Modifications

• Introduce a new flow util similar to FanOut, but balances the work across go funcs. I feel like this could be useful for parallelizing work across cpu cores.

Verify change

• [x] Make sure the change passes the CI checks.

Solve the memory leak problem caused by the equality of sw.size and sw.slide

Motivation

• Fixes #___

Modifications

Verify change

• [ ] Make sure the change passes the CI checks.

Motivation

• Fix memory leak casused by time.After()

Modifications

• Use time.NewTicker() outside the loop

Verify change

• [ ] Make sure the change passes the CI checks.

I test the fanout function in flow_test.go, it seems not correct when I add another flow. var toReture = func(in interface{}) interface{} { msg := in.(string) return msg }

func TestFlowUtil(t *testing.T) { in := make(chan interface{}) out := make(chan interface{}) source := ext.NewChanSource(in) flow1 := flow.NewMap(toUpper, 1) flow2 := flow.NewMap(toReture, 1) filter := flow.NewFilter(filterA, 1) sink := ext.NewChanSink(out) var _input = []string{"a", "b", "c"} var _expectedOutput = []string{"b", "b", "c", "c"} go ingest(_input, in) go deferClose(in, time.Second*10) go func() { fanOut := flow.FanOut(source.Via(filter).Via(flow1).Via(flow2, 2) fanOut[1].To(sink) }() var _output []string for e := range sink.Out { _output = append(_output, e.(string)) } sort.Strings(_output) assertEqual(t, _expectedOutput, _output) } The function 'toReture' does nothing ,but it return incorrect result.

Motivation

• Added ZipWith functionality, e.g. taking elements from several upstreams in a round-robin fashion, bundling them to a single element, then sending it downstream.

Modifications

• Implementation is in utils.go, the test is in flow_test.go.

Verify change

• [V] Make sure the change passes the CI checks.

Hi,

I was looking at row 133 of kafka_sarama.go source code and noticed that every consumed message are pushed to the same channel. If I'm correct, this have the effect to make the processing of the messages to be sequential, loosing the possibility of processing the partitions in parallel. Could it make sense to have a channel per partition in this case?

Motivation

• feat: add keyBy、connect, coFlatMap for a stream join a stream

Modifications

Verify change

• [ ] Make sure the change passes the CI checks.

Is there a way to support grouping data by time and length？ like:

	ch := make(chan any, 5)
	go func() {
		for i := 0; i < 10000; i++ {
			time.Sleep(rndTime)
			ch <- i
		}
	}()
	source := extension.NewChanSource(ch)

	m := flow.NewMap(func(i interface{}) interface{} {
		logrus.Infof("%v", i)
		return i
	}, 1)
        // not supported now
	th := flow.NewBufferWithTimeOrCount(time.Millisecond * 100, 50) // every 100 mil secs or 50 counts emits data

	source.Via(th).Via(m).To(extension.NewIgnoreSink())

expection print:

[x,x,x,x...x,x,x] // 50 counts
// if 100 mil secs pass
[x,x,...x,x] // data counts less than 50

There are many instances running in Distributed architecture. Some status are shared between instances, like time windows. We could put it on Redis, or other storage.