module Control.Bitcoin.Api.Transaction where
import qualified Data.Conduit as C (Source)
import Control.Concurrent (forkIO, killThread,
myThreadId, threadDelay)
import Control.Concurrent.STM.TBMQueue (isClosedTBMQueue, newTBMQueue,
writeTBMQueue)
import Control.Lens ((^.))
import Control.Monad (unless)
import Control.Monad.IO.Class (liftIO)
import Control.Monad.STM (atomically)
import Data.Conduit.TQueue (sourceTBMQueue)
import qualified Data.Bitcoin.Block as Btc
import qualified Data.Bitcoin.Transaction as Btc
import qualified Network.Bitcoin.Api.Blockchain as Blockchain
import qualified Network.Bitcoin.Api.Types as T
watch :: T.Client
-> Maybe Integer
-> C.Source IO Btc.Transaction
watch client Nothing = watch client (Just 6)
watch client (Just confirmations) = do
chan <- liftIO $ atomically $ newTBMQueue 16
curHeight <- liftIO blockHeight
_ <- liftIO $ forkIO $ watchNext chan curHeight
sourceTBMQueue chan
where
blockHeight = do
limit <- Blockchain.getBlockCount client
return (limit confirmations)
watchNext chan height = do
cur <- blockHeight
if cur > height
then go chan (height + 1)
else threadDelay 1000000 >> watchNext chan height
go chan height = do
block <- Blockchain.getBlock client =<< Blockchain.getBlockHash client height
tid <- myThreadId
result <- mapM (insert chan) (block ^. Btc.blockTxns)
let isClosed = False `elem` result
if isClosed
then killThread tid
else watchNext chan height
insert chan tx = atomically $ do
isClosed <- isClosedTBMQueue chan
unless isClosed (writeTBMQueue chan tx)
return isClosed