{-# LANGUAGE DeriveDataTypeable #-}

-----------------------------------------------------------------------
-- |
-- Module     : Control/Concurrent/Chan.hs
-- Copyright  : (c) Ivan Tomac 2012
-- License    : BSD3
--
-- Maintainer : ivan `dot` tomac `at` google `dot` com
-- Stability  : experimental
--
-- Unbounded channels. Provided for compatibility with the old API.
--
-- A channel is implemented using a blocking concurrent 'Queue' and
-- three locks, one for reading, one for writing and one for modifying
-- the queue.
--
-----------------------------------------------------------------------

-----------------------------------------------------------------------
--  read:
--      read lock
--          q lock
--              grab token
--          blocking read
--          q lock
--              release token
--      return value
--
--  write:
--      write lock
--          write value
--
--  duplicate:
--      write lock
--          duplicate channel
--      return new channel
--
--  unget:
--      write lock
--          q lock
--              if q empty write
--              if token missing skip one value and unget
--              if token present unget
--
--  isempty:
--      write lock
--          q lock
--              is q empty
-----------------------------------------------------------------------

module Control.Concurrent.Chan
  ( -- * The 'Chan' type
    Chan

    -- * Operations
  , newChan
  , writeChan
  , readChan
  , dupChan
  , unGetChan
  , isEmptyChan

    -- * Stream interface
  , getChanContents
  , writeList2Chan
  ) where

import Control.Applicative
import Control.Arrow
import Control.Concurrent.MVar
import Data.IORef
import Data.Maybe
import Data.Typeable
import System.IO.Unsafe

import Control.Concurrent.Queue.Blocking

-- | 'Chan' is an abstract type representing an unbounded FIFO channel.
data Chan a = Chan (MVar (Queue a, Bool)) (MVar (Queue a)) (MVar ())
    deriving Typeable

-- | Returns a new instance of 'Chan'.
newChan :: IO (Chan a)
newChan = do q <- newQ'
             Chan <$> newMVar (q, False) <*> newMVar q <*> newMVar ()

-- | Writes a value to a 'Chan'.
writeChan :: Chan a -> a -> IO ()
writeChan (Chan _ vw _) x = modifyMVar_ vw (`unsafeWriteQ` x)

-- | Reads the next value from a 'Chan'.
readChan :: Chan a -> IO a
readChan (Chan vq _ vr) = withMVar vr $ const $ do
    (x, q) <- lock vq >>= readBlockingQ
    unlock vq q
    return x
    where
    lock   v   = modifyMVar  v $ return . (flip (,) True >>= (,)) . fst
    unlock v q = modifyMVar_ v $ const  $ return (q, False)

-- | Duplicates a 'Chan'. The duplicate channel begins empty, but data
-- written to either channel from then on will be available on both.
-- Hence this creates a kind of broadcast channel, where data written
-- by anyone is seen by everyone else.
dupChan :: Chan a -> IO (Chan a)
dupChan (Chan _ vw _) = modifyMVar vw dup
    where dup q = do ch <- (`Chan` vw) <$> newMVar (q, False)
                                       <*> newMVar ()
                     return (q, ch)

-- | Put an item back into a channel, where it will be the next item
-- read.
unGetChan :: Chan a -> a -> IO ()
unGetChan (Chan vq vw _) x = modifyMVar_ vw $ withMVar vq . unget
    where
    unget qw (qr, b)
      | qw == qr  = unsafeWriteQ qw x
      | otherwise = do
            Q r <- (if b then skipOne else return) qr
            readIORef r >>= either
                (const $ unsafeWriteQ qw x)
                ((>> return qw) . (`modifyMVar_` update))

    skipOne     = fmap (snd . fromJust) . readQ
    update next = (,) x . Q <$> (newMVar next >>= newIORef . Right)

-- | Returns 'True' if the 'Chan' is empty.
isEmptyChan :: Chan a -> IO Bool
isEmptyChan (Chan vq vw _) =
    withMVar vw $ withMVar vq . fmap return . (. fst) . (==)

-- | Returns a lazy list representing the contents of the 'Chan', much
-- like 'System.IO.hGetContents'.
getChanContents :: Chan a -> IO [a]
getChanContents = liftA2 (:) <$> (unsafeInterleaveIO . readChan)
                             <*> getChanContents

-- | Write an entire list of items to a 'Chan'.
writeList2Chan :: Chan a -> [a] -> IO ()
writeList2Chan = mapM_ . writeChan