module ExampleBroadcast where

import           Control.Applicative
import           Control.Monad
import           Control.Concurrent (threadDelay)
import           MVC
import           MVC.Prelude as MVC
import qualified Pipes.Network.TCP as TCP
import Data.ByteString (ByteString)
import Control.Concurrent.Async
import Control.Concurrent.STM

newtype Broadcast a = Broadcast { unBroadcast :: TVar (Output a) }

broadcast :: STM (Output a, Broadcast a)
broadcast = do
    tvar <- newTVar mempty
    let output = Output $ \a -> do
            o <- readTVar tvar
            send o a
    return (output, Broadcast tvar)
{-# INLINABLE broadcast #-}

subscribe :: Buffer a -> Broadcast a -> IO (Input a)
subscribe buffer (Broadcast tvar) = do
    (output, input) <- spawn buffer
    atomically $ modifyTVar' tvar (mappend output)
    return input
{-# INLINABLE subscribe #-}

newtype Funnel a = Funnel { unFunnnel :: TVar (Input a) }

funnel :: STM (Funnel a, Input a)
funnel = do
    tvar <- newTVar mempty
    let input = Input $ do
            i <- readTVar tvar
            recv i
    return (Funnel tvar, input)
{-# INLINABLE funnel #-}

widen :: Buffer a -> Funnel a -> IO (Output a)
widen buffer (Funnel tvar) = do
    (output, input) <- spawn buffer
    atomically $ modifyTVar' tvar (mappend input)
    return output
{-# INLINABLE widen #-}

broadcastServer :: (String, String) -> Managed (View ByteString, Controller ByteString)
broadcastServer (h,p) = join $ managed $ \k -> do
    -- make a Broadcast channel
    (outB, refB, refF, inF) <- atomically $ do
        (outB', refB') <- broadcast
        (refF', inF') <- funnel
        return (outB', refB', refF', inF')
    let server = TCP.listen (TCP.Host h) p $ \(sock,_) ->
            forever $
            TCP.acceptFork sock $ \(sock',_) -> do
                inClient <- subscribe Unbounded refB
                outClient <- widen Unbounded refF
                aIn <- async $ runEffect $
                       for (fromInput inClient) (lift . TCP.send sock')
                aOut <- async $ runEffect $
                        TCP.fromSocket sock' 4096 >-> toOutput outClient
                void $ waitAnyCancel [aIn, aOut]
    let v = asSink (void . atomically . MVC.send outB)
        c = asInput inF
    withAsync server $ \_ -> k ((,) <$> pure v <*> (pure c <> keepOpenController))

keepOpenController :: Managed (Controller a)
keepOpenController = MVC.producer Single $ do
    lift $ threadDelay (365 * 24 * 60 * 60 * 10^6)
    return ()

