FChan.hs

{-# LANGUAGE Trustworthy, DeriveDataTypeable #-}

-- | Functional channels
-- | A channel data type which allows consumers to hold references to different points in a stream at the same time. Elements of a channel are kept alive only so long as there are references pointing before those elements. And producers on a channel are kept alive only so long as there are consumers.
module Control.CUtils.FChan (Chan, listToChan, chanContents, DoneReadingException(..), takeChan, tryTakeChan, newChan, makeConsumer, dupChan) where

import Control.Concurrent.MVar
import Control.Concurrent (forkIO)
import Control.Monad
import Control.Exception
import System.Mem.Weak
import Data.Typeable
import Data.IORef

import System.IO.Unsafe

newtype Chan t = Chan {-# NOUNPACK #-} (MVar (t, Chan t))

-- | Construct a channel from a list.
{-# NOINLINE listToChan #-}
listToChan :: [t] -> Chan t
listToChan (x:xs) = Chan (unsafePerformIO (newMVar (x, listToChan xs)))
listToChan [] = Chan (unsafePerformIO newEmptyMVar)
-- Referential transparency is preserved because a means of adding
-- to a channel is not available unless explicitly provided.

-- | Thrown by the writer function.
data DoneReadingException = DoneReadingException deriving (Typeable, Show)

instance Exception DoneReadingException

addChan :: MVar (Chan t) -> t -> IO ()
addChan vr x = modifyMVar_ vr (\chn -> do
	may <- return (Just chn)
	case may of
		Just (Chan vr2) -> do
			vr' <- newEmptyMVar
			let chn' = Chan vr'
			putMVar vr2 (x, chn')
			-- mkWeak chn' chn' Nothing
			return chn'
		Nothing -> throwIO DoneReadingException)

-- | Take the first element from a channel, and a channel representing the remainder of the output.
takeChan (Chan vr) = readMVar vr

tryTakeChan (Chan vr) = tryReadMVar vr

-- | Create a new channel. The first return value is a function that can be used to add values to the channel. The second return value is the channel itself.
newChan = do
	vr <- newEmptyMVar
	vr2 <- newEmptyMVar
	let chn = Chan vr
	-- weak <- mkWeak chn chn Nothing
	putMVar vr2 chn
	return (addChan vr2, chn)

-- | The first return value is a thunk that returns values from the channel successively, starting from the position of the parameter channel. The second thunk can be used to retrieve the position of the channel after all the reads made using the first thunk.
makeConsumer chn = do
	vr2 <- newMVar chn
	return (modifyMVar vr2 (\chn -> do
		(x, chn2) <- takeChan chn
		return (chn2, x)),
		readMVar vr2)

chanContents :: Chan t -> IO [t]
chanContents chn = tryTakeChan chn >>= maybe
	(return [])
	(\(x, xs) -> liftM (x:) (chanContents xs))

-- | Create a channel which is initially empty, but accumulates new elements.
dupChan chn = tryTakeChan chn >>= maybe
	(return chn)
	(dupChan . snd)