Wie wird aus einem Pull-basierten Rohr ein Push-basiertes Rohr gemacht?

Question

Standardmäßig sind Pipes Pull-basiert. Dies ist auf den Operator zurückzuführen, der über +>> implementiert wird, der der sinnvolle Operator bind für seine Pull-Kategorie ist. Mein Verständnis ist, dass dies bedeutet, dass, wenn Sie Code wie producer >-> consumer haben, der Körper des Verbrauchers zuerst genannt wird, dann, sobald es Daten erwartet, wird der Produzent angerufen.

Ich habe in der pipes Dokumentation gesehen here, dass Sie den Code (reflect .) von Pipes.Core können einen Zug auf Basis Rohr in ein Push-basierten Rohr zu drehen. Das heißt stattdessen (korrigiere mich, wenn ich falsch liege), dass im Code über producer >-> consumer der Produzent zuerst ausgeführt wird, einen Wert produziert, dann versucht der Konsument zu konsumieren. Das scheint wirklich nützlich zu sein und ich würde gerne wissen, wie es geht.

Ich habe auch in den Diskussionen here gesehen, dass es kein Push-basiertes Gegenstück zu gibt, weil es einfach ist, jedes Rohr herum zu drehen (ich nehme an mit?), Aber ich kann nicht wirklich herausfinden, wie man es macht oder finde irgendwelche Beispiele.

Hier einige Code, den ich versucht habe:

stdin :: Producer String IO r 
stdin = forever $ do 
    lift $ putStrLn "stdin" 
    str <- lift getLine 
    yield str 

countLetters :: Consumer String IO r 
countLetters = forever $ do 
    lift $ putStrLn "countLetters" 
    str <- await 
    lift . putStrLn . show . length $ str 

-- this works in pull mode 
runEffect (stdin >-> countLetters) 

-- equivalent to above, works 
runEffect ((\() -> stdin) +>> countLetters) 

-- push based operator, doesn't do what I hoped 
runEffect (stdin >>~ (\_ -> countLetters)) 

-- does not compile 
runEffect (countLetters >>~ (\() -> stdin)) 

Answer 1

-- push based operator, doesn't do what I hoped 
runEffect (stdin >>~ (\_ -> countLetters)) 

ich das Problem hier zu sammeln ist, dass, während der Produzent lief zunächst wie erwartet, der erste erzeugte Wert fallen gelassen wird. Vergleichen ...

GHCi> runEffect (stdin >-> countLetters) 
countLetters 
stdin 
foo 
3 
countLetters 
stdin 
glub 
4 
countLetters 
stdin 

... mit:

GHCi> runEffect (stdin >>~ (\_ -> countLetters)) 
stdin 
foo 
countLetters 
stdin 
glub 
4 
countLetters 
stdin 

Dieses Problem wird im Detail durch Gabriel Gonzalez‘Antwort auf this question diskutiert. Es läuft darauf hinaus, wie das Argument zu der Funktion, die Sie an (>>~) geben, der "treibende" Eingang in dem Push-basierten Fluss ist, und so, wenn Sie const es entfernt Sie am Ende der ersten Eingabe. Die Lösung ist countLetters entsprechend neu zu gestalten:

countLettersPush :: String -> Consumer String IO r 
countLettersPush str = do 
    lift $ putStrLn "countLetters" 
    lift . putStrLn . show . length $ str 
    str' <- await 
    countLettersPush str' 

GHCi> runEffect (stdin >>~ countLettersPush) 
stdin 
foo 
countLetters 
3 
stdin 
glub 
countLetters 
4 
stdin 

ich auch in Diskussionen gesehen habe here, dass es kein Push-Pendant zu >->, weil es einfach ist, jedes Rohr herum zu drehen (I vermute mit reflect?)

Ich bin mir meines Grundes nicht ganz sicher, aber es scheint, dass das nicht ganz auf die obige Lösung zutrifft.Was wir können jetzt tun, dass wir die Push-basierten Flow haben ordnungsgemäß funktioniert, reflect verwendet, um sie um zurück zu einem Pull-basierten Flow:

-- Preliminary step: switching to '(>~>)'. 
stdin >>~ countLettersPush 
(const stdin >~> countLettersPush)() 

-- Applying 'reflect', as the documentation suggests. 
reflect . (const stdin >~> countLettersPush) 
reflect . const stdin <+< reflect . countLettersPush 
const (reflect stdin) <+< reflect . countLettersPush 

-- Rewriting in terms of '(+>>)'. 
(reflect . countLettersPush >+> const (reflect stdin))() 
reflect . countLettersPush +>> reflect stdin 

Diese in der Tat Pull-basiert, wie die Strömung wird durch reflect stdin die stromabwärtige Client angetrieben:

GHCi> :t reflect stdin 
reflect stdin :: Proxy String()() X IO r 
GHCi> :t reflect stdin :: Client String() IO r 
reflect stdin :: Client String() IO r :: Client String() IO r 

die Strömung, beinhaltet jedoch String s Upstream-Senden, und so kann es nicht in Bezug auf (>->) ausgedrückt werden, das ist sozusagen stromabwärts-only:

GHCi> -- Compare the type of the second argument with that of 'reflect stdin' 
GHCi> :t (>->) 
(>->) 
    :: Monad m => 
    Proxy a' a() b m r -> Proxy() b c' c m r -> Proxy a' a c' c m