Come rilevare la fine dell'ingresso con i tubi

Question

Sto provando a leggere un gruppo di massimo 50 elementi da una pipe e elaborarli in un'azione IO tutto in una volta. (Il caso d'uso per questo è che sto cercando di inserire i dati in un database e voglio fare un intero batch all'interno di una transazione perché è molto più efficiente). Ecco una versione semplificata di quello che ho ottenuto finora:

type ExampleType = Int 

doSomething :: [ExampleType] -> IO() 
doSomething = undefined 

inGroupsOf50 :: Monad m => Producer ExampleType m() -> m() 
inGroupsOf50 input = 
    runEffect $ input >-> loop 
     where loop = do entries <- replicateM 50 await 
        lift $ doSomething entries --Insert a bunch all in one transaction 
        loop 

Il problema è per quanto posso dire, a meno che il numero di elementi da inserire capita di dividere per 50, ho intenzione di perdere alcuni. Quello che voglio veramente invece di replicateM 50 await è qualcosa che mi dà fino a 50 voci o meno se l'input finisce, ma non riesco a capire come scriverlo.

Ho pensato che pipes-parse potrebbe essere la libreria giusta da guardare. draw sembra avere una firma promettente ... ma finora tutti i bit non si incastrano nella mia testa. Ho un producer, sto scrivendo uno consumer e non ho capito come si riferisce al concetto di parser.

Answer 1

Anche più di pipes-parse si consiglia di dare un'occhiata a pipes-group. In particolare, esaminiamo la funzione

-- this type is slightly specialized 
chunksOf 
    :: Monad m => 
    Int -> 
    Lens' (Producer a m x) (FreeT (Producer a m) m x) 

Il bit Lens' è forse spaventoso, ma può essere eliminata rapidamente: si afferma che siamo in grado di convertire Producer a m x-FreeT (Producer a m) m x [0]

import Control.Lens (view) 

chunkIt :: Monad m => Int -> Producer a m x -> FreeT (Producer a m) m x 
chunkIt n = view (chunksOf n) 

Così ora dobbiamo capire cosa fare con quel FreeT bit. In particolare, stiamo andando a voler scavare nel pacchetto free ed estrarre la funzione iterT

iterT 
    :: (Functor f, Monad m) => 
    (f (m a) -> m a) -> 
    (FreeT f m a -> m a) 

Questa funzione, iterT, andiamo noi "consumano" un FreeT uno "step" in un momento. Per capire questo, faremo prima specializzati il ​​tipo di iterT per sostituire f con Producer a m

runChunk :: Monad m => 
      (Producer a m (m x)  -> m x) -> 
      (FreeT (Producer a m) m x -> m x) 
runChunk = iterT 

In particolare, runChunk in grado di "run" un FreeT piena di Producer s fino a quando diciamo che come convertire un Producer a m (m x) in un m -azione. Questo potrebbe iniziare a sembrare più familiare. Quando definiamo il primo argomento di runChunk, dobbiamo semplicemente eseguire un Producer che, in questo caso, non avrà più del numero selezionato di elementi.

Ma cosa succede con il valore restituito efficace m x? È la "continuazione", ad es. tutti i blocchi che vengono dopo il quello corrente che si sta scrivendo. Così, per esempio, supponiamo che abbiamo un Producer di Char s e vorremmo stampare e linea dopo 3 caratteri

main :: IO() 
main = flip runChunk (chunkIt 3 input) $ \p -> _ 

Il _ buco, a questo punto è di tipo IO() con p nel contesto in digitare p :: Producer Char IO (IO()). Possiamo consumare questa pipe con for, raccogliere il suo tipo restituito (che è la continuazione, di nuovo), emettere una nuova riga e quindi eseguire la continuazione.

input :: Monad m => Producer Char m() 
input = each "abcdefghijklmnopqrstuvwxyz" 

main :: IO() 
main = flip runChunk (chunkIt 3 input) $ \p -> do 
    cont <- runEffect $ for p (lift . putChar) 
    putChar '\n' 
    cont 

E questo comporta esattamente come desiderato

λ> main 
abc 
def 
ghi 
jkl 
mno 
pqr 
stu 
vwx 
yz 

---

Per essere chiari, mentre ho fatto un po 'di esposizione, questo è il codice abbastanza semplice una volta si vede come tutti i pezzi si incastrano. Ecco l'elenco completo:

input :: Monad m => Producer Char m() 
input = each "abcdefghijklmnopqrstuvwxyz" 

main :: IO() 
main = flip iterT (input ^. chunksOf 3) $ \p -> do 
    cont <- runEffect $ for p $ \c -> do 
    lift (putChar c) 
    putChar '\n' 
    cont 

[0] Anche un po 'di più, ma è sufficiente per ora.