Because Clojure doesn't check for a lot of things, GIGO.

One of my favorite examples:

(clojure.set/intersection #{1 2} [2 3])
=> #{1}

If you write user/foo.bar, you get the value and no syntax error. I tested in clj whose version is:

$ clj --version
Clojure CLI version 1.11.1.1429

^ that doesn't mean it should be used or is supposed to work that way.

What is the official role of dots in names?

By the way, quoting dotted symbols neither works:

`aaa
user/aaa

`foo.bar
foo.bar

> '.' has special meaning - it can be used one or more times in the middle of a symbol to designate a fully-qualified class name, e.g. java.util.BitSet, or in namespace names.

@U06BDSLBVC6 BTW that version is the CLI version, not version of Clojure itself. But doesn't really matter here.

https://clojurians.slack.com/archives/C03S1KBA2/p1711976053093099?thread_ts=1711975547.236739&cid=C03S1KBA2 This is the version of Clojure which is installed in my PC. Projects are, of course, free to require their own Clojure versions in their deps.edn or project.clj files.

That is the version of the Clojure CLI that's installed on your PC. The Clojure-the-language version will be different. Clojure-the-language doesn't even use commit-level versions. It's just three digits and an optional qualifier, something like 1.11.2 or 1.12.0-alpha3. This is how you can check the Clojure version that's used by your CLI on your particular machine (that can have settings that alter that version):

$ clj -M -e '(clojure-version)'
"1.11.1"

I ran your code:

$ clj -M -e '(clojure-version)'
"1.11.1"

As I said, that can easily be changed on a global level. So by itself clj --version doesn't tell you the version of the language that will be used.

$ file deps.edn
deps.edn: cannot open `deps.edn' (No such file or directory)
$ clj --version
Clojure CLI version 1.11.1.1435
$ clj -M -e '(clojure-version)'
"1.9.0"

The three first parts of the CLI version do tell you the Clojure version that will be used by default, if there are no other configurations and no potential stale cache issues (no clue whether it's something relevant ATM but I'm pretty sure in the past there used to be some bugs with cp caching). But if you give someone else that version without any accompanying information, they cannot robustly infer the Clojure version that's been used.

(~/clojure)-(!2002)-> clj --version
Clojure CLI version 1.11.2.1446

Mon Apr 01 12:29:56
(~/clojure)-(!2003)-> clj -M -e '(clojure-version)'
"1.12.0-alpha9"

That sounds very vague so I can't give any concrete advice. Maybe core.async will be of use to you, channels let you implement throttling. And the overall description of the problem sounds cohesive with what core.async offers.

ok, any pointers in how they can be fit together in a declarative way like in my pseudocode example? https://clojure.org/guides/async_walkthrough This jumps from a very surface overview straight to API docs (which tend to be a bit obscure) and source code

or should I just read through all the core.async functions? 🙂

or is the magic in the <???> part to connect the parts using something like pipeline?

There should be some guides on core.async, but I'm not familiar with them so can't suggest anything concrete. The API has functions that let you provide your own transducers. But throughput/backpressure is something else. I think I saw an article once showing how to implement a throttler on top of a channel in a few lines, but I don't recall where.

Ok thank you, I will look for more resources about core.async

My only doubt is that I have the impression that transducers only work synchronously but maybe that’s not the case

Transducers are a way to create a reducing function, that's it. That function can then be used in any context. Of course, there's a caveat about stateful transducers.

I might be confusing them with the simple cases of map and filter for example

For some reason I took your throttling to mean "batch by ms" and did this. If you just mean "drop repeated puts in that time span" you can maybe just use a dropping buffer

(defn throttle [ms out in]
  (a/go-loop [batch []
              timeout (a/timeout ms)]
    (a/alt!
      timeout ([_]
               (a/onto-chan! out batch false)
               (recur []
                      (a/timeout ms)))
      in ([x]
          (recur (conj batch x)
                 timeout))))
  out)


(def in (a/chan 10
          (comp (filter even?)
                (map str))))
(def out (a/chan (a/sliding-buffer 10)))

(throttle 1000 out in)

(dotimes [x 10]
  (a/put! in (rand-int 10)))
(a/poll! out)

Oh, found this in my notes: https://github.com/brunoV/throttler

If you enter that URL into the search bar here, you'll find multiple discussions with other approaches and library suggestions.

You can also check out pipeline functions in core async to parallelize the transducers https://clojuredocs.org/clojure.core.async/pipeline

alright, thank you both 🙂 I will look into your suggestions, appreciate it

@U064UGEUQ my main difference from your example is that I want to hold messages to be released later in some cases. I wanted to implement this using transducers so that they could compose neatly with filter and map, but I'm not seeing a way to do this since I need something like an external go-loop that has access to the output channel, and that channel is only available inside the call to the transducer function returned from my throttle transducer

What do you mean by: > hold messages to be released later in some cases What cases exactly?

ok, so my messages have a key and timestamp, and I want to release them independently per key at a target update rate. So if my target update rate is for example 5s, if a message comes for a particular key that is <5s after the previous one, I want to hold it, and release it later if no messages arrive until the 5s expire for that key

if multiple messages come within the 5s window, I will only keep the latest one and release it when 5s have elapsed since the last message released for that key

if a new message arrives that is >5s after the last one, it is immediately released (and stored in a hashmap for timestamp comparison with the next one)

The https://github.com/cgrand/xforms library should have some useful stuff. I think you can use a combination of its x/by-key and x/window-by-time transducers to help get what you're looking for, I think.

or x/reductions if all else fails

alright, I'll take a look, thanks

Though I wouldn't rule out just trying it with a go loop either, sounds like you could code it up pretty directly as one

yea, that was what I was trying but I don't see how I can access the out channel inside the go-loop without explicitly creating it and passing it around, which will make me lose the ability to neatly compose throttle with other transducers (I think?)

If you decide to create a transducer, you must not use core.async. Not "cannot" but "must not". Because go blocks use threads from a very limited thread pool. Randomly waiting in such threads will exhaust the pool and bring your whole app to a halt.

If you want any abstraction mechanism and not necessarily transducers, you can build it on top of core.async with your own functions. Most functions will create transducers and call pipeline-blocking. Some functions, like throttle, will use pipeline-async.

yes I just read that somewhere 😁 I'm still trying out stuff so that's why I'm using go but I will take that into account

My interpretation was that he's not waiting on stuff, just "caching" messages by some id and then evicting them after some amount of time has elapsed

I'm building this stuff to be used by non-clojure developers in a Java interop context, so I want to get the usability/interface of it as simple as possible. That's why I wanted to make it similar/work seamlessly with filter and map when describing a sequence of operations to be performed on the data

You can still do it. Your namespace can have two functions, add-transducer and add-throttling.

ok, but sounds like it's not trivial or desirable to treat them the same inside my coordination/composition internals, i.e. the user doesn't have to know or care what is or isn't a transducer

(<???>
       (filter ...)
       (map ...)
       (throttle 1000 ...)
       (map ...))

this is the interface I would like

You can do it by making (throttle ...) return some wrapper that's not a function and then use it to call pipeline-async.

so I would just dispatch on that inside <???> ?

Yes.

I have something like this in mind, haven't tested though:

(defn throttle [ms]
  {:async-handler (fn [value chan]
                    ;; Do whatever needs to be done to put
                    ;; `value` onto `chan` at the right time.
                    )})

(defn build-pipeline [source n
                      & steps]
  (let [bundled-steps (into []
                            (comp (partition-by fn?)
                                  (mapcat (fn [bundle]
                                            (if (fn? (first bundle))
                                              [(apply comp bundle)]
                                              bundle))))
                            steps)]
    (reduce (fn [from step]
              (let [to (a/chan)]
                (if (fn? step)
                  (a/pipeline-blocking n to step from)
                  (a/pipeline-async n to {:async-handler step} from))
                to))
            source bundled-steps)))

(let [source (a/chan)
      dest (build-pipeline source 10 
                           (map inc)
                           (filter even?)
                           (partition-all 3)
                           (throttle 1000))]
  ...)

ok I think I understand this better now, thank you both

what is the difference between pipeline-blocking and pipeline btw?

No difference.

I remember a particular arcane longwinded way of demunging or something that someone once mentioned ages ago. I remember laughing at it considering it as being insane.

No way to do it without using the var itself. Either by using #' (or var) or a macro. Demunging might be somewhat useful only for things like functions, such as in your example. But not for vars in general since e.g. 7 is a plain number, there's nothing to munge.

Yeah, not vars in general. In my case it is a fn.

Ooo, a macro.

Oh, BTW - in your example, that metadata is not on the var but on the args vector. So you can't access it at all.

Yeah made a mistake. I forgot to remove the caret.

Things that often begin with “ooo macro” don’t often end well.

Thanks anyway. I’ll find a way around

One thing worth noting is that metadata might be a poor fit for what you want to do anyway. But, of course, it depends. Maybe it is a good fit. Maybe an explicit registry is a better fit. Maybe some introspection. And so on.

I’m trying to imitate some system someone else has built using Java annotations(ugh). They have all this annotated data that they use to make runtime decisions.

You’re right about the registry, which is sort of what I’d shown them originally. Just wanted to bring it closer to their idiom I suppose. Not a big deal

A custom function-defining macro is much closer to Java annotations than plain metadata. :)

Actually, I’d prefer something like reitit or malli.

But yeah, a macro system for sure.

maybe https://clojuredocs.org/clojure.repl/demunge

which I think can be used like https://github.com/clojure/spec.alpha/blob/c630a0b8f1f47275e1a476dcdf77507316bad5bc/src/main/clojure/clojure/spec/alpha.clj#L131

caveat emptor, ymmv, here be dragons, etc.

You never close in the first case - the output isn't flushed, I would assume. Try the same but with exiting the process.

Or with calling (.flush the-writer) explicitly (`doto` is helpful here). But in any case, you should of course be using with-open if a writer has a well-defined scope and you don't need to also read from that file at the same time.

Oh yeah! Flush, I missed that in my mental model for sure I was curious why first it was not working. Thanks for clarifying it for me

When I created a BufferedWriter, the bytes were written to the buffer and not transmitted to the file system.