<commit_before>#!/usr/bin/env runhaskell
{-# LANGUAGE OverloadedStrings
           , ScopedTypeVariables
  #-}
module Database.PostgreSQL.Pulse where

import           Control.Applicative
import           Control.Concurrent
import           Control.Concurrent.STM (STM)
import qualified Control.Concurrent.STM as STM
import qualified Control.Concurrent.STM.TChan as STM
import           Control.Monad
import           System.IO

import           Data.ByteString (ByteString)
import qualified Data.ByteString as ByteString hiding (hPutStrLn)
import qualified Data.ByteString.Char8 as ByteString (lines, hPutStrLn)
import qualified Data.Csv as Cassava
import qualified Data.Csv.Incremental as Cassava
import           Data.Vector (Vector)
import qualified Data.Vector as Vector


main :: IO ()
main  = do (a :: STM.TChan ByteString)          <- STM.atomically $ STM.newTChan
           (b :: STM.TChan (Vector ByteString)) <- STM.atomically $ STM.newTChan
           (c :: STM.TChan [Vector ByteString]) <- STM.atomically $ STM.newTChan
           _ <- forkIO . forever $ getBlock a
           _ <- forkIO . forever . STM.atomically $ breakIntoRecords a b
           _ <- forkIO . forever $ send c
           forever $ delayedBulkTransfer 25000 b c

delayedBulkTransfer :: Int -> STM.TChan t -> STM.TChan [t] -> IO ()
delayedBulkTransfer micros from to = do
  (forkIO . STM.atomically) (STM.writeTChan to =<< readAll from)
  threadDelay micros

readAll :: STM.TChan t -> STM [t]
readAll chan = do h <- STM.tryReadTChan chan
                  case h of Just h  -> (h:) <$> readAll chan
                            Nothing -> return []

breakIntoRecords :: STM.TChan ByteString
                 -> STM.TChan (Vector ByteString) -> STM () 
breakIntoRecords i o = do
  records <- Vector.fromList . ByteString.lines <$> STM.readTChan i
  STM.writeTChan o records

getBlock :: STM.TChan ByteString -> IO ()
getBlock o = do
  bytes <- ByteString.hGet stdin (2^20)
  hPutStrLn stderr "Got a block."
  when (bytes /= "") $ STM.atomically (STM.writeTChan o bytes)

send :: STM.TChan [Vector ByteString] -> IO ()
send i = do
  vecs <- STM.atomically $ STM.readTChan i
  let lists = (Vector.toList <$> vecs)
  mapM_ (mapM_ (ByteString.hPutStrLn stdout)) lists

<commit_msg>Testing this way works well.<commit_after>#!/usr/bin/env runhaskell
{-# LANGUAGE OverloadedStrings
           , ScopedTypeVariables
  #-}
module Database.PostgreSQL.Pulse where

import           Control.Applicative
import           Control.Concurrent
import           Control.Concurrent.STM (STM)
import qualified Control.Concurrent.STM as STM
import qualified Control.Concurrent.STM.TChan as STM
import           Control.Monad
import           System.IO

import           Data.ByteString (ByteString)
import qualified Data.ByteString as ByteString hiding (hPutStrLn)
import qualified Data.ByteString.Char8 as ByteString (lines, hPutStrLn)
import qualified Data.Csv as Cassava
import qualified Data.Csv.Incremental as Cassava
import           Data.Vector (Vector)
import qualified Data.Vector as Vector


main :: IO ()
main  = do (a :: STM.TChan ByteString)          <- STM.atomically $ STM.newTChan
           (b :: STM.TChan (Vector ByteString)) <- STM.atomically $ STM.newTChan
           (c :: STM.TChan [Vector ByteString]) <- STM.atomically $ STM.newTChan
           _ <- forkIO . forever $ getBlock a
           _ <- forkIO . forever . STM.atomically $ breakIntoRecords a b
           _ <- forkIO . forever $ send c
           forever $ delayedBulkTransfer 500000 b c

delayedBulkTransfer :: Int -> STM.TChan t -> STM.TChan [t] -> IO ()
delayedBulkTransfer micros from to = do
  (forkIO . STM.atomically) (STM.writeTChan to =<< readAll from)
  threadDelay micros

readAll :: STM.TChan t -> STM [t]
readAll chan = do h <- STM.tryReadTChan chan
                  case h of Just h  -> (h:) <$> readAll chan
                            Nothing -> return []

breakIntoRecords :: STM.TChan ByteString
                 -> STM.TChan (Vector ByteString) -> STM () 
breakIntoRecords i o = do
  records <- Vector.fromList . ByteString.lines <$> STM.readTChan i
  STM.writeTChan o records

getBlock :: STM.TChan ByteString -> IO ()
getBlock o = do
  bytes <- ByteString.hGet stdin (16)
  hPutStrLn stderr "Got a block."
  when (bytes /= "") $ STM.atomically (STM.writeTChan o bytes)

send :: STM.TChan [Vector ByteString] -> IO ()
send i = do
  vecs <- STM.atomically $ STM.readTChan i
  let lists = (Vector.toList <$> vecs)
  mapM_ (mapM_ (ByteString.hPutStrLn stdout)) lists
  hFlush stdout


{'fraction_non_alphanumeric': 0.10026385224274406, 'fraction_numerical': 0.00395778364116095, 'mean_word_length': 2.6626409017713364, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5421048', 'n_tokens_mistral': 1529, 'n_tokens_neox': 1380, 'n_words': 420}