As I read through the source code, I see a lot of the following

(t, d) => {
  if (t == 0){
    ...
  } else if (t == 1){
    ...
  } else {
    ...
  }
}

I've typed up a quick helper to create these callbags and was wondering if you think that it would be useful more widely.

const build = cbs => (t, d) => {
  const arr = cbs instanceof Array ? cbs : [ cbs.start, cbs.data, cbs.end ];
  arr[t] && arr[t](d)
}

This allows you to create a callbag passing either an object with {start, data, end} functions or an array of functions.

build({
  start(d){
    ...
  },
  data(d){
    ...
  },
  end(d){
    ...
  }
});

build([
  d => ...,
  d => ...,
  d => ...
])

const interval = period => build({
  start(sink){
    let i = 0;
    const id = setInterval(() => sink(1, i++), period);
    sink(0, build({ end: () => clearInterval(id) }));
  }
})

I'd be interested to hear what feedback you have. Thank you for your time and for your novel thoughts on Observables and AsyncIterables!

If I was using a callbag pipeline in React, say, in my componentDidMount, how would I tell the pipeline to stop listening for events on a componentWillUnmount?

Howdy! Would you be open to a pull request that changes your build tooling to output distributable packages for multiple targets?

As is, it's problematic to include this module in a project targeted for the browser due to use of ES6+ features (arrow functions, spread,...). I think it's not uncommon for build tooling targeting the browser to exclude ./node_modules/ from transpilation (Babel). Of course... the user could always adjust their build, but.... Also, the module built here doesn't support tree shaking to reduce bundle size.

Current version of callbag-from-iter:

https://github.com/staltz/callbag-from-iter/blob/44a38236a13c5dd72a2a0783aa2abb5db7b8bb8d/package.json#L3

fromIter's terminate handling is added in version 1.2.0

Even though it was bug fixing, callbag-basics still uses version 1.0.x of callbag-from-iter.

we should update this.

A debounce callbag function would be useful.

From Callbag Spec:

Every consumer of data is a function (type: number, payload?: any) => void

However, the forEach factory returns a function of different signature:

const forEach = operation => source => {...}

So strictly speaking, it is not a callbag factory?

Following the build it yourself philosophy of this library, I made my own lib of callbags.
Just a package.json with the dependencies I need and an index.js exporting them like the one in this project.

However I noticed callbag-flatten failed to work, because the new version doesn't export the function like the rest of the callbags. Now it needs to be
flatten: require('callbag-flatten').default,
instead of the previous
flatten: require('callbag-flatten')

I wonder, shouldn't all calbags export the same way?

The goal of one of my callbag streams is to insert each processed item into a database. How can my final sink (forEach) be asynchronous? Why can't any callbag-basics handler also be asynchronous?

Hi André !

I just found out about your spec. I find it very nice and I love the idea of a new angle on that topic. I have been playing around with it last night and built my own operators ( a partition mostly ). When I get the time I'll do a rewrite with tests and may well publish it.

Back on the topic. I have read your article about repos without maintainers. There as well I really like your philosophy and I tend to read more and more open source code. I have to say I've been reading the rxjs implementation of partition before I gave it a go. So, I don't want to bother you too much, but maybe you could give me a direction. I'd like to implement an unsubscribe operator. I know I should use the type 2 for that, but I'm not sure how to tackle it.

Love the idea! It's simple and elegant.

I wanted to write debounce to make sure I understood the philosophy of callbag. Before I published it, I was wondering if you could just take a quick look and make sure it's written correctly. My testing seems to work, but this is totally new to me.

const debounce = duration => source => (start, sink) => {
  if (start !== 0) return;
  let latest = 0;
  let talkback;
  source(0, (t, d) => {
    if (t === 0) {
      talkback = d;
      sink(t, d);
    } else if (t === 1) {
      const now = Date.now();
      if (now >= latest + duration) {
        latest = now;
        sink(t, d);
      } else {
        talkback(1);
      }
    } else {
      sink(t, d);
    }
  });
};

From what I observed you follow semantic versioning in all your callbag-* packages. Since patch and minor releases therefore should not contain any breaking changes, can we pin all deps in callbag-basics to their major versions?

    "callbag": "1.0.x",
    "callbag-for-each": "1.0.x",
    "callbag-from-obs": "1.1.x",
    ...

==>

    "callbag": "1",
    "callbag-for-each": "1",
    "callbag-from-obs": "1",
    ...

I can make a PR for this if you agree.

Is it not desirable to propagate error emitted from the sink back to the source?

https://github.com/staltz/callbag-flatten/blob/9d08c8807802243517697dd7401a9d5d2ba69c24/index.js#L7-L10

Up for discussion, but would the Typescript community benefit from stricter types?
It saves a few headaches when your compiler can hold your hand a bit more :)

A few things I would be looking for:

• Allow the error type to be specified
• Stricter signal / parameters for each type of callbag

Here is something I have been using:

export enum Signal {
  START = 0,
  DATA = 1,
  END = 2,
}

export type TalkbackArgs = [signal: Signal.DATA] | [signal: Signal.END]
export type Talkback = (...op: TalkbackArgs) => void

export type SinkArgs<E, A> =
  | [signal: Signal.START, talkback: Talkback]
  | [signal: Signal.DATA, data: A]
  | [signal: Signal.END, error?: E]
export type Sink<E, A> = (...op: SinkArgs<E, A>) => void

export type SourceArgs<E, A> = [signal: Signal.START, sink: Sink<E, A>]
export type Source<E, A> = (...op: SourceArgs<E, A>) => void

export type Callbag<E, A> = Source<E, A> | Sink<E, A>

callbag-basics ought to have a very simple technique to introduce/push values onto a stream. The ability to emit any value onto the stream manually is more important than fromEvent, fromPromise, fromIterator, and fromObservable. Consider that all of these "from" modules do not belong in callbag-basics. Consider that callbag-basics ought to have a single basic module for introducing values onto a stream.

Good idea by the way

Typescript thinks the expected operators are exported, but at runtime, 7 of the exports are Modules.
Using the latest version of Next.js, which bundles with webpack.

A centralized repo (from which all your packages are published) might be a better fit for this.

The first example from Try it online section - Observe Events doesn't seem to work:

(0 , _callbagBasics.fromEvent) is not a function

  16 | const setText = event => (prev, curr) => event.target.value;
> 17 | const inputs = pipe(fromEvent(input, "input"), map(setText));
     |                              ^
  18 | 

alas, I don't know how to fix this.

Thanks!

Hey @staltz.

I like the stuff you build, but I am wondering what will happen if there is an error in one of the source factories. Observables or similar libraries like xstream always have a value handler and an error handler for that reason.

Maybe you can elaborate that a bit in the README.

Cheers

When I invoke callbag-flatten, I expect that it will flatten an array. But instead I see errors that source and innerSource "is not a function". How is callbag-flatten supposed to work? With the Node stream API, it's pretty simple to push new values into a stream.

pipe(
	interval(1),
	map((each) => ([each, each])),
	flatten,
	forEach(function(each) {
		console.log('each: ' + JSON.stringify(each));
	})
);

Hi,

In RxJS we can write something like this : Observable.of({}).subscribe(...). I need to build a similar operator for callbags but I can't seem to get it to work.

I have tried a naive implementation which is :

export const fromObject = obj => (start, sink) => {
	if (start !== 0) return;

	sink(0, (t) => {});
	sink(1, obj);
	sink(2);
};

So the idea is, I greet the sink, I send him the object, and I terminate it. The problem arises with merge() which gives me a "talkback is not a function". After looking at the code of merge, it seems that the issue comes from that line : if (++startCount === n) sink(0, talkback);. When it's looped on, the talkbacks are not set yet at the time I send my sink(1, obj). I'm not saying there's something wrong with the merge operator, but I look for guidance on how to tackle the problem and build my source.

The goal is to have a synchronous source that emits once and completes. And that plays nicely with the current ecosystem.