clojure
core.async

Operaciones básicas del canal: crear, poner, tomar, cerrar y buffers.

core.async se trata de hacer procesos que tomen valores y pongan valores en canales .

(require [clojure.core.async :as a])

Creando canales con chan

Se crea un canal utilizando la función chan :

(def chan-0 (a/chan)) ;; unbuffered channel: acts as a rendez-vous point.
(def chan-1 (a/chan 3)) ;; channel with a buffer of size 3. 
(def chan-2 (a/chan (a/dropping-buffer 3)) ;; channel with a *dropping* buffer of size 3
(def chan-3 (a/chan (a/sliding-buffer 3)) ;; channel with a *sliding* buffer of size 3

Poniendo valores en canales con >!! y >!

Pones valores en un canal con >!! :

(a/>!! my-channel :an-item)

Puede poner cualquier valor (cadenas, números, mapas, colecciones, objetos, incluso otros canales, etc.) en un canal, excepto nil :

;; WON'T WORK
(a/>!! my-channel nil)
=> IllegalArgumentException Can't put nil on channel

Dependiendo del buffer del canal, >!! puede bloquear el hilo actual.

(let [ch (a/chan)] ;; unbuffered channel
  (a/>!! ch :item) 
  ;; the above call blocks, until another process 
  ;; takes the item from the channel.
  )
(let [ch (a/chan 3)] ;; channel with 3-size buffer
  (a/>!! ch :item-1) ;; => true
  (a/>!! ch :item-2) ;; => true
  (a/>!! ch :item-3) ;; => true
  (a/>!! ch :item-4) 
  ;; now the buffer is full; blocks until :item-1 is taken from ch.
  )

Desde el interior de un bloque (go ...) , puede y debe usar a/>! en lugar de a/>!! :

 (a/go (a/>! ch :item))

El comportamiento lógico será el mismo que a/>!! , pero solo se bloqueará el proceso lógico de goroutine en lugar del hilo del sistema operativo real.

Usando a/>!! dentro de un bloque (go ...) hay un anti-patrón:

;; NEVER DO THIS
(a/go 
  (a/>!! ch :item))

Tomando valores de canales con <!!

Toma un valor de un canal usando <!! :

;; creating a channel
(def ch (a/chan 3))
;; putting some items in it
(do 
  (a/>!! ch :item-1)
  (a/>!! ch :item-2)
  (a/>!! ch :item-3))
;; taking a value
(a/<!! ch) ;; => :item-1
(a/<!! ch) ;; => :item-2

Si no hay ningún elemento disponible en el canal, a/<!! bloqueará el hilo actual hasta que se coloque un valor en el canal (o se cierre el canal, ver más adelante):

(def ch (a/chan))
(a/<!! ch) ;; blocks until another process puts something into ch or closes it

Desde el interior de un bloque (go ...) , puede y debe usar a/<! en lugar de a/<!! :

 (a/go (let [x (a/<! ch)] ...))

El comportamiento lógico será el mismo que a/<!! , pero solo se bloqueará el proceso lógico de goroutine en lugar del hilo del sistema operativo real.

Usando a/<!! dentro de un bloque (go ...) hay un anti-patrón:

;; NEVER DO THIS
(a/go 
  (a/<!! ch))

Canales de cierre

Cierra un canal con a/close! :

(a/close! ch)

Una vez que se cierra un canal, y se han agotado todos los datos en el canal, las tomas siempre devolverán nil :

(def ch (a/chan 5))

;; putting 2 values in the channel, then closing it
(a/>!! ch :item-1)
(a/>!! ch :item-2)
(a/close! ch)

;; taking from ch will return the items that were put in it, then nil
(a/<!! ch) ;; => :item-1
(a/<!! ch) ;; => :item-2
(a/<!! ch) ;; => nil
(a/<!! ch) ;; => nil
(a/<!! ch) ;; => nil

;; once the channel is closed, >!! will have no effect on the channel:
(a/>!! ch :item-3)
=> false ;; false means the put did not succeed
(a/<!! ch) ;; => nil

Asíncrono pone con put!

Como alternativa a a/>!! (que puede bloquear), puedes llamar a/put! para poner un valor en un canal en otro hilo, con una devolución de llamada opcional.

(a/put! ch :item)
(a/put! ch :item (fn once-put [closed?] ...)) ;; callback function, will receive 

En ClojureScript, ya que no es posible bloquear el hilo actual, a/>!! No es compatible, y put! es la única forma de colocar datos en un canal desde fuera de un bloque (go) .

Toma asíncrona con take!

Como alternativa a a/<!! (que puede bloquear el hilo actual), puede usar a/take! para tomar un valor de un canal de forma asíncrona, pasándolo a una devolución de llamada.

(a/take! ch (fn [x] (do something with x)))

Utilizando buffers de caída y deslizamiento

Con los buffers que caen y se deslizan, los bloques nunca se bloquean, sin embargo, cuando el buffer está lleno, se pierden datos. Al eliminar el búfer se pierden los últimos datos agregados, mientras que los búferes deslizantes pierden los primeros datos agregados.

Ejemplo de búfer de caída:

(def ch (a/chan (a/dropping-buffer 2)))

;; putting more items than buffer size
(a/>!! ch :item-1)
=> true ;; put succeeded
(a/>!! ch :item-2)
=> true
(a/>!! ch :item-3)
=> false ;; put failed

;; no we take from the channel
(a/<!! ch)
=> :item-1
(a/<!! ch)
=> :item-2
(a/<!! ch)
;; blocks! :item-3 is lost

Ejemplo de búfer deslizante:

(def ch (a/chan (a/sliding-buffer 2)))

;; putting more items than buffer size
(a/>!! ch :item-1)
=> true
(a/>!! ch :item-2)
=> true
(a/>!! ch :item-3)
=> true

;; no when we take from the channel:
(a/<!! ch)
=> :item-2
(a/<!! ch)
=> :item-3
;; :item-1 was lost