Files accompanying blog post on wjwh.eu

Main.hs

{-# LANGUAGE OverloadedStrings, DeriveGeneric, DerivingStrategies, DeriveAnyClass #-}
module Main where

import Prelude hiding (foldl)
import Conduit hiding (connect)
import Data.Attoparsec.ByteString as Attoparsec
import Data.Attoparsec.ByteString.Char8
import qualified Data.ByteString as B
import Data.Aeson
import Data.Conduit.Attoparsec
import Data.Conduit.Combinators
import Data.Conduit.Network
import Data.Maybe
import Control.Monad
import qualified Data.ByteString.Char8 as C
import GHC.Generics
import Network.Socket
import Network.Socket.ByteString (recv, sendAll)

data CounterMessage = CounterMessage { counter :: Integer, time :: Integer }
    deriving stock (Show, Eq, Generic)
    deriving anyclass (ToJSON, FromJSON)

-- main :: IO ()
-- main = do
--   let settings = clientSettings 8000 "127.0.0.1"
--   runTCPClient settings $ \ad -> do
--     runConduitRes $ appSource ad .| stdout

singleLine :: Parser Value
singleLine = skipSpace *> json' <* char '\n'

-- main :: IO ()
-- main = do
--   let settings = clientSettings 8000 "127.0.0.1"
--   runTCPClient settings $ \ad -> do
--     runConduitRes $ appSource ad
--                  .| conduitParser singleLine
--                  .| mapC snd
--                  .| printC

-- Object (fromList [("counter",Number 1.0),("time",Number 1.646592622e9)])
-- Object (fromList [("counter",Number 2.0),("time",Number 1.646592623e9)])
-- Object (fromList [("counter",Number 3.0),("time",Number 1.646592624e9)])
-- Object (fromList [("counter",Number 4.0),("time",Number 1.646592625e9)])

-- main :: IO ()
-- main = do
--   let settings = clientSettings 8000 "127.0.0.1"
--   runTCPClient settings $ \ad -> do
--     runConduitRes $ appSource ad
--                  .| linesUnboundedAscii
--                  .| mapC (decodeStrict :: C.ByteString -> Maybe CounterMessage)
--                  .| filterC isJust
--                  .| mapC fromJust
--                  .| printC

-- CounterMessage {counter = 1, time = 1646593912}
-- CounterMessage {counter = 2, time = 1646593913}
-- CounterMessage {counter = 3, time = 1646593914}
-- CounterMessage {counter = 4, time = 1646593915}

data CounterState = CounterState { total :: Integer, lastUpdate :: Integer } deriving (Show,Eq)

updateCounterState :: CounterState -> CounterMessage -> CounterState
updateCounterState (CounterState total _) (CounterMessage counter newTime) = CounterState (total + counter) newTime



-- main :: IO ()
-- main = do
--   let settings = clientSettings 8000 "127.0.0.1"
--   runTCPClient settings $ \ad -> do
--     runConduitRes $ appSource ad
--                  .| linesUnboundedAscii
--                  .| mapC (decodeStrict :: C.ByteString -> Maybe CounterMessage)
--                  .| filterC isJust
--                  .| mapC fromJust
--                  .| scanlC updateCounterState (CounterState 0 0)
--                  .| printC

-- CounterState {total = 0, lastUpdate = 0}
-- CounterState {total = 1, lastUpdate = 1646596859}
-- CounterState {total = 3, lastUpdate = 1646596860}
-- CounterState {total = 6, lastUpdate = 1646596861}
-- CounterState {total = 10, lastUpdate = 1646596862}

main :: IO ()
main = do
  let settings = clientSettings 8000 "127.0.0.1"
  runTCPClient settings $ \ad -> do
    runConduitRes $ appSource ad
                 .| linesUnboundedAscii
                 .| mapC (decodeStrict :: C.ByteString -> Maybe CounterMessage)
                 .| filterC isJust
                 .| mapC fromJust
                 .| scanlC updateCounterState (CounterState 0 0)
                 .| iterM (liftIO . Prelude.print)
                 .| filterC (odd . total)
                 .| mapC (C.pack . (++ "\n") . show . total)
                 .| appSink ad

socketstreamer.rb

require 'socket'
require 'json'

server = TCPServer.new 8000
client = server.accept    # Wait for a client to connect
counter = 0

Thread.new do
  loop do
    puts client.gets
    sleep 1
  end
end

loop do
  counter += 1
  payload = { counter: counter, time: Time.now.to_i }.to_json
  client.puts(payload)
  sleep 1
end