blob: 4a311277e93e821c13bdb106ae1caaec8f460d57 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ViewPatterns #-}
{-# LANGUAGE TupleSections #-}
{-# LANGUAGE StandaloneDeriving #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
{-# LANGUAGE DeriveGeneric #-}
{-# LANGUAGE BangPatterns #-}
module KikiD.Multiplex where
import System.IO
import qualified Data.ByteString.Char8 as B
import Data.Monoid
import Control.Concurrent.STM
import Data.Map.Strict as M
import Control.Monad
import Control.Concurrent
import qualified Data.Binary as Bin
import Control.Concurrent.STM.TBMQueue
import Control.Monad.Loops
import Data.List
import Data.Maybe
-- | pipeTransHookMicroseconds
--
-- This function indefinitely reads the @fromChan@ queue and applies
-- the function @translate@ to the contents before passing it on to the
-- @toChan@ queue. The @triggerAction@ is performed on the message prior
-- to the translation. The @fromChan@ queue is checked every @micros@
-- microseconds from the last emptying.
--
-- To terminate the thread, close @fromChan@ queue.
--
pipeTransHookMicroseconds :: TBMQueue a -> TBMQueue b -> Int -> (x -> a -> Maybe b) -> (a -> IO x) -> IO ()
pipeTransHookMicroseconds fromChan toChan micros translate triggerAction =
whileM_ (fmap not (atomically $ isClosedTBMQueue fromChan)) $ do
whileM_ (fmap not (atomically $ isEmptyTBMQueue fromChan)) $ do
msg <- atomically $ readTBMQueue fromChan
case msg of
Just m' -> do
x <- triggerAction m'
case translate x m' of
Just m -> atomically $ writeTBMQueue toChan m
_ -> return ()
_ -> return ()
threadDelay micros -- 5000 -- yield two 100ths of a second, for other threads
pipeTransHook fromChan toChan translate triggerAction =
pipeTransHookMicroseconds fromChan toChan 5000 translate triggerAction
pipeTrans fromChan toChan translate =
pipeTransHook fromChan toChan translate (void . return)
pipeHook fromChan toChan triggerAction =
pipeTransHook fromChan toChan id triggerAction
pipeQueue fromChan toChan =
pipeTransHookMicroseconds fromChan toChan 5000 (\() -> Just) (void . return)
teePipeQueueMicroseconds fromChan toChan1 toChan2 micros =
whileM_ (fmap not (atomically $ isClosedTBMQueue fromChan)) $ do
whileM_ (fmap not (atomically $ isEmptyTBMQueue fromChan)) $ do
msg <- atomically $ readTBMQueue fromChan
case msg of
Just m -> do
atomically $ writeTBMQueue toChan1 m
atomically $ writeTBMQueue toChan2 m
_ -> return ()
threadDelay micros -- 5000 -- yield two 100ths of a second, for other threads
teePipeQueue fromChan toChan1 toChan2 =
teePipeQueueMicroseconds fromChan toChan1 toChan2 5000
-- Deprecated: Use consumeQueueMicroseconds
-- TODO: Remove this
withQueueMicroseconds fromChan action delay = whileM_ (atomically . fmap not $ isClosedTBMQueue fromChan) $ do
whileM_ (atomically . fmap not $ isEmptyTBMQueue fromChan) $ do
t <- atomically $ readTBMQueue fromChan
case t of
Just x -> action x
Nothing -> return ()
threadDelay delay
{-# ANN withQueue ("HLint: Ignore Eta reduce"::String) #-}
withQueue fromchan action = consumeQueueMicroseconds fromchan 5000 action
{-# DEPRECATED withQueueMicroseconds, withQueue "Use consumeQueueMicroseconds" #-}
-- | consumeQueueMicroseconds
-- (as of version 1.0.4)
--
-- Continously run the provided action on items
-- from the provided queue. Delay for provided
-- microseconds each time the queue is emptied.
consumeQueueMicroseconds q micros action = whileM_ (atomically . fmap not $ isClosedTBMQueue q) $ do
whileM_ (atomically . fmap not $ isEmptyTBMQueue q) $ do
x <- atomically $ readTBMQueue q
case x of
Just s -> action s
Nothing -> return ()
threadDelay micros
|
