Added new buffer type 'New'

Like 'Latest' but with a guaranteed write between any two reads.

Conflicts:
	src/Pipes/Concurrent.hs

src/Pipes/Concurrent.hs

@@ -47,7 +47,7 @@ module Pipes.Concurrent (
     ) where
 
 import Control.Applicative (
-    Alternative(empty, (<|>)), Applicative(pure, (<*>)), (<*), (<$>) )
+    Alternative(empty, (<|>)), Applicative(pure, (*>), (<*>)), (<*), (<$>) )
 import Control.Concurrent (forkIO)
 import Control.Concurrent.STM (atomically, STM)
 import qualified Control.Concurrent.STM as S
@@ -184,6 +184,9 @@ spawn' buffer = do
         Latest a  -> do
             t <- S.newTVarIO a
             return (S.writeTVar t, S.readTVar t)
+        New       -> do
+            m <- S.newEmptyTMVarIO
+            return (\x -> S.tryTakeTMVar m *> S.putTMVar m x, S.takeTMVar m)
 
     sealed <- S.newTVarIO False
     let seal = S.writeTVar sealed True
@@ -226,6 +229,7 @@ data Buffer a
         'Latest' is never empty nor full.
     -}
     | Latest a
+    | New
 
 {- $reexport
     @Control.Concurrent@ re-exports 'forkIO', although I recommend using the

tests/tests-main.hs

@@ -114,9 +114,11 @@ main = do
     runTest (testSenderClose $ Bounded 7) "BoundedNotFilledSenderClose"
     runTest (testSenderClose Single) "SingleSenderClose"
     runTestExpectTimeout (testSenderCloseDelayedSend $ Latest 42) "LatestSenderClose"
+    runTest (testSenderCloseDelayedSend New) "NewSenderClose"
     --
     runTest (testReceiverClose Unbounded) "UnboundedReceiverClose"
     runTest (testReceiverClose $ Bounded 3) "BoundedFilledReceiverClose"
     runTest (testReceiverClose $ Bounded 7) "BoundedNotFilledReceiverClose"
     runTest (testReceiverClose Single) "SingleReceiverClose"
     runTest (testReceiverCloseDelayedReceive $ Latest 42) "LatestReceiverClose"
+    runTest (testReceiverClose New) "NewReceiverClose"