the core.async docs lists a bunch of deprecated functions, like map< and filter<, saying "Use transducer instead". But there doesn't seem to be any function that takes a channel and applies a transducer to it. What have I missed?

@magnars: (chan buf xf) will apply the xf to everything that passes through the channel

yes, but that conflates creating the channel and applying a transducer to it - which fails in my case, since I'm not the one creating the channel

fair enough. In that case you may need to splice it in yourself

can use pipe with a channel of your own creation to get the transducer application?

yeah, that is certainly doable - sounds like something that would fit into core.async core tho

(let [c (chan 1 (map :message))] (pipe ws-channel c) c) isn't exactly the epitome of elegance

useful as it can be, core.async isn’t exactly the epitome of elegance

true that, but this conflation of creation and transformation seems to be at odds with Clojure's notion of simplicity

that’s actually what i’m referring to. Although there’s ways of mitigating some of the pain, core.async is really the first time I’ve felt like a core Clojure library embraced imperative APIs, mutability, and identity. It’s a great tool, but it feels distinctly different from so much of the rest of Clojure.

well, core.async channels are stateful. so, there’s that

The point of channels is conveyance... processes are inherently harder

I see it half-glass-full in that channels help simplify subsystems

+1!

certainly better than writing callbacks

my intention was not to point out flaws inherent in core.async, just a wish for a function that applies a transducer to an existing channel

there's no way to do that without a race condition

sorry, let me rephrase - returns a new channel with the transducer applied

I basically want to do (map< my-fn my-chan) with transducers, something like (<xf (map my-fn) my-chan) maybe.

Yeah, I think make a new channel and pipe is your best bet

yeah, one problem with (let [c (chan 1 (map my-fn))] (pipe my-chan c) c) is that it'll take five minutes out of my video to explain it, while the deprecated map< would not. It also seems to me like a good fit for a basic feature in core.async.

it seems like such a basic thing to do, that I'm surprised it's not there

¯\(ツ)/¯

libraries usually prefer finding primitives...

rather than conveniences

wouldn't "decomplecting" creation and transformation be simpler to work with, in a non-convenient sense?

I don't follow

let me try again creating a channel is a primitive operation. In my mind, applying a transducer to a channel is a primitive operation too. Please note, this would of course not change the original channel, but return a new one, just like the rest of clojure works. It's obvious the core.async-creators thought along the same lines when they created map< and filter< etc. Then transducers came along, with a deprecation "please use transducers" - but those were bolted on the chan constructor, complecting together channel creation and transformation.

so if you want to, say filter a channel, you have to create it yourself.

you dismissed this by saying that what I was asking for was just a convenience (easy, in Hickey-parlance), but my point is that it would be conceptually simpler.

Our codebase uses core.async in a lot of places, and we’ve written a lot of higher-level functions similar in spirit to the one you would have to write to get this behavior.

magnars: what happens when you apply a transducer to a channel with a transducer already?

magnars: not saying we'll never change the API, but especially with async things there's a lot of subtleties

I'm not saying "apply the transducer to the channel itself"

ah, I must be missing something.

can you post some pseudo-code?

I guess what I'm asking is, could something like this be part of the core.async API:

(defn <xf [ch xf]
  (let [c (chan 1 xf)]
    (pipe ch c)
    c))

possibly implemented in a less naive fashion

as it stand, map< has been deprecated with no real replacement

it probably won't be replaced

merely because you can do it just like the code you posted

(`map<` had other bad issues too regarding exceptions)

so you don't think "applying a transducer to a channel" is a primitive operation?

no

I'm gonna need some time to wrap my head around that, but fair enough

it's different than seqs in core because channels aren't datastructures

there's no notion of persistence

but channels are first class citizens, in that they can be passed around - and now transducers are only added at the point of creation

I can't get a "firehose" channel, and then filter it based on some criteria

without going through hoops, that feel to me highly stateful, with pipe

but channels have state

Btw, pipe returns you to channel so you don't need the let (pipe my-ch (chan 1 (map inc)))

maybe I'm thinking about channels too much like "Rx"

https://clojuredocs.org/clojure.core.async/filter%3C

rauh: pipe's return signature may change in the future to return a channel that closes when the piping process is done

pkobrien: that is also subsumed by transducer and will disappear

or is filter also deprecated...

@meow: yeah, that's the one I want, except it would take the transducer as an argument, but it's deprecated and not getting a replacement

@ghadi: ok, yeah

but I've only been working with core.async a few months, and am possibly thinking badly about it

I'm thinking about it as a stream of events, and I'd like to filter/map those events

but doing those transformations does not feel supported in the core.async API today

well, its easy to put a transducer on the channel that you create, but you know that

except I'm not the one creating the channel

or some other part of the code could be doing that

right

using pipe the way @rauh mentioned looks OK, tho, but the return value of pipe is apparently not part of the official API atm

Yeah but you could easily re-do pipe that still behaves that way in case it does change.

sure, I could also easily add the <xf function I pasted earlier but since this seems "obvious" to me, but not the people working on core.async, I'm left with the feeling that there is some central idea that I've not grasped yet. Which is what got me here in the first place.

and after this, I'm left with a vague notion - something like this: since channels are stateful, they don't compose as well as data structures

also, using a channel someone else (some entirely other part of the code) created isn't done very often

@magnars: so you've got one part of the program creating a channel that is to be passed to another part of your program for the other part to do puts onto the channel, yes?

I'm still learning the best way to use core.async myself, so just thinking aloud in case it helps.

in my case, I'd just like to (map< :message my-ch) before passing the channel to the consumer of the channel, to remove knowledge of the entire datastructure to the consumer.

then I noticed map< was deprecated, went to look for the replacement, found only chan, which I could not use, since I didn't create my-ch in the first place

I then started using pipe - but was still curious why my intuitions about channels were so wrong

ie, why didn't the maintainers of core.async think creating a channel and filtering a channel were two separate concerns, simpler apart than together

One thing I'm curious about is what is the main benefit of using a transducer as opposed to simply passing a value to some function before putting it on a channel since you can't have transducers on the way out, and in the wild it seems like a lot of the examples I've seen have been just that - function calls on the way in or out, rather than transducers.

it's a composability-issue - maybe I have a channel with lots of events on it. Before I pass it to you, I want to filter in only the events relevant to, say, a single user.

that might be the domain of the auth-code - and should be kept there

I'm not sure I understand you completely. What do you mean with "found only chan, which I could not use, since I didn't create my-ch in the first place"?

Think of channels simply as FIFO buffers that you interact with, you can't have different "views" on them. You'll need a new channel

chan was the only place to "add" a transducer

meaning you can specify a transducer when you create a channel

The thing with anything that modifies a channel is: You need a new channel for this anyways. Look at the map< implementation. It also created a new channel.

yeah, like I'm saying: I don't want to modify a channel in-place, that would be even more un-Clojure-like

but I see that transducer when you create the channel as just sort of a convenience, since like rauh says they are just FIFO buffers

I'd just like a way to say "hey, map this channel with this function" in a way that feels natural and supported

But then what do you want the return value to be?

Imagine if you had to map over a vector in clojure.core with something like (let [c (chan 1 (map my-fn))] (pipe my-chan c) c)

the new channel

I basically want this in the API:

(defn <xf [ch xf]
  (let [c (chan 1 xf)]
    (pipe ch c)
    c))

Oh I think I got you

I guess you'd have to create that convenience API yourself

indeed - and that's fine. But it seemed like such a basic thing to be doing, that I was surprised it was not in the API.

which led me on this quest in this channel to understand why it is in fact NOT a basic thing to be doing.

because I'm making a video, and I'd like to use core.async in an idiomatic way

Well one thing is that you gain very little by this convenience function and then you'll have the argument "if this function made it into core.async then I also want function xyz"

again, in my mind, the functions map< filter< distinct< etc were remove since they were "redundant", and transducers came to the rescue

but there is no plan to actually give us the transducer-version of these functions

Which is often an argument that some convenience function don't make it into core libraries, to prevent getting tons of other functions that do very little

E.g. the chan version adds buffers which are more powerful. You can do the usual >1 buffer size or dropping buffers etc

I realize it is my own limited understanding here, but it sounds to me like you're saying "map and filter are only convenience functions, because you can do both of those with reduce, and we don't want to go on a slippery slope"

and that is true, in a way, but you'd always have "map" and "filter" anyway, because they are core - and in my understanding, so would "give me a new channel, with this channel as input, transduced like this" be - core.

but I don't want to waste more of your time on this. Thanks for your patience. I'll show myself out. 😛

I guess it could be cool to add another arity to chan and allow a from-chan to make things nicer, but chan already has so many arguments ;/

Or (chan from-chan (map inc)) would be cool and consider a from-chan to be a "form of buffer"

@magnars: But I doubt any of this is going to happen.

yeah, I'm not hoping for change - I'm hoping for insight. At this point, I've got two points of insight out of this: 1) map, reduce, filter might be basic operations on event streams, but not for channels. The core.async team does not think about channels as event streams, even if I do. I think here is the mismatch. 2) my ability to articulate an idea in textual form is sorely lacking.

what about (clojure.core.async/map :message [my-ch])?

magnar: speaking for myself, I think channels are more generic than event streams (encompassing alt! and other coordination)

you can do event-streamy things, but a lot more