How to turn a thrust-based pipe into a push?

Question

By default, pipes are pull based. This is due to the >-> operator, which is implemented via +>> , which is the bind point operator for its traction category. My understanding is that this means that if you have code like producer >-> consumer , the producer >-> consumer body will be called first, and then, as soon as it expects the data, the producer will be called.

I saw pipes in the documentation here so that you can use the code (reflect .) From Pipes.Core to include an ejector pipe in the current based pipe. This means instead (correct me if I am mistaken) that in the code above producer >-> consumer , the manufacturer starts first, returns a value, then the consumer tries to consume. This seems really useful, and I would like to know how to do it.

I also saw in the discussions here that there is no exact analogue >-> , because it’s easy to turn any channel around (I assume with reflection?), But I can’t figure out how to do this or find examples.

Here is the code I tried:

 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)) 

I understand that the problem is that although the manufacturer starts first, as expected, the first produced value is discarded. Compare ...

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

... from:

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

This question is discussed in detail by Gabriel Gonzalez in response to this question . This boils down to how the argument of the function you give (>>~) is the "control" input in the stream, so if you are const , you end up discarding the first input. The solution is to modify countLetters :

 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 

I also saw in the discussions here that there is no exact analogue >-> , because it is easy to turn any channel around (I assume with reflection?)

I am not entirely sure of my position, but it seems that this is not entirely applicable to the solution above. What can we do now that the stream stream is working correctly, we use reflect to rotate it back to pulling based on the stream:

 -- 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 

It really depends on the stream, as the stream is controlled by reflect stdin , below Client :

 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 

However, the stream involves sending the String upstream, and therefore it cannot be expressed in terms of (>->) , that is, so to speak, only with the flow:

 GHCi> -- Compare the type of the second argument with that of 'reflect stdin' GHCi> :t (>->) (>->) :: Monad m => Proxy a' a () bmr -> Proxy () bc' cmr -> Proxy a' ac' cm