socketed (empty) → 0.1.0.0
raw patch · 6 files changed
+348/−0 lines, 6 filesdep +MissingHdep +asyncdep +base
Dependencies added: MissingH, async, base, bytestring, conduit-combinators, conduit-extra, http-types, optparse-applicative, stm, stm-chans, stm-conduit, template-haskell, text, wai, wai-websockets, warp, websockets
Files
- LICENSE +20/−0
- Main.hs +33/−0
- Network/Socketed.hs +72/−0
- Network/Socketed/Internal.hs +74/−0
- Network/Socketed/Template.hs +83/−0
- socketed.cabal +66/−0
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2017 sen.cenan@gmail.com++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be included+in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ Main.hs view
@@ -0,0 +1,33 @@+module Main where++import Data.Semigroup ((<>))+import qualified Options.Applicative as Opt++import Network.Socketed (SocketedOptions(..), runSocketedServer)++params :: Opt.Parser SocketedOptions+params = SocketedOptions+ <$> Opt.strOption+ ( Opt.long "bind"+ <> Opt.short 'b'+ <> Opt.help "host ip the server will be bind to"+ <> Opt.showDefault+ <> Opt.value "0.0.0.0"+ <> Opt.metavar "STRING" )+ <*> Opt.option Opt.auto+ ( Opt.long "port"+ <> Opt.short 'p'+ <> Opt.help "port number the server will be running on"+ <> Opt.showDefault+ <> Opt.value 3000+ <> Opt.metavar "INT" )++main :: IO ()+main = Opt.execParser opts >>= runSocketedServer+ where+ opts = Opt.info (Opt.helper <*> params)+ (+ Opt.fullDesc+ <> Opt.progDesc "socketed"+ <> Opt.header "socketed"+ )
+ Network/Socketed.hs view
@@ -0,0 +1,72 @@+module Network.Socketed (+ SocketedOptions(..), runSocketedServer+ ) where++import Conduit (+ ConduitM, MonadIO,+ (.|), mapM_C, runConduit, sinkList+ )+import Data.Conduit.TMChan (TMChan, sinkTMChan, dupTMChan, sourceTMChan)++import Control.Concurrent.Async (async)+import Control.Concurrent.STM (atomically)+import Control.Concurrent.STM.TMChan (newBroadcastTMChanIO)++import Data.ByteString (ByteString)+import Data.ByteString.Char8 (pack, unpack)+import Data.ByteString.Lazy (fromStrict)+import Data.Void (Void)++import Network.HTTP.Types (status200)+import Network.Wai (Application, responseLBS)+import Network.Wai.Handler.Warp (+ Settings,+ defaultSettings, setPort, setHost, runSettings+ )+import Network.Wai.Handler.WebSockets (websocketsOr)++import Network.WebSockets (+ ServerApp,+ acceptRequest, defaultConnectionOptions, sendTextData+ )++import Network.Socketed.Internal (+ SocketedOptions(..),+ stdinLines, showWSHost, takeUntil2Empty+ )+import Network.Socketed.Template (evalHtml)++sinkStdinToChan :: MonadIO m => TMChan ByteString -> ConduitM a Void m ()+sinkStdinToChan = (stdinLines .|) . flip sinkTMChan False++serverSettings :: SocketedOptions -> Settings+serverSettings (SocketedOptions h p) = setHost (read hoststr)+ $ setPort p defaultSettings where hoststr = "Host \"" ++ h ++ "\""++socketedApp :: [ByteString] -> ByteString -> TMChan ByteString -> Application+socketedApp rls html broadcastChan+ = websocketsOr defaultConnectionOptions wsApp backupApp where+ wsApp :: ServerApp+ wsApp pendingConn = do+ conn <- acceptRequest pendingConn+ chan <- atomically $ dupTMChan broadcastChan+ _ <- mapM (sendTextData conn) rls -- send replayed lines+ runConduit $ sourceTMChan chan .| mapM_C (sendTextData conn)++ backupApp :: Application+ backupApp _ respond = respond+ $ responseLBS status200 [] (fromStrict html)++runSocketedServer :: SocketedOptions -> IO ()+runSocketedServer opts@(SocketedOptions h p) = do+ putStrLn "Accepting replayed data: "+ rls <- runConduit $ stdinLines .| takeUntil2Empty .| sinkList+ putStrLn "Replayed data: "+ mapM_ print rls+ putStrLn $ "\nStart streaming @ " ++ showWSHost h p+ chan <- newBroadcastTMChanIO+ mirror <- atomically $ dupTMChan chan+ _ <- async . runConduit $ sinkStdinToChan chan+ _ <- async . runConduit $ sourceTMChan mirror .| mapM_C (putStrLn . unpack)+ let app = socketedApp rls (pack $ evalHtml (length rls) opts) chan+ runSettings (serverSettings opts) app
+ Network/Socketed/Internal.hs view
@@ -0,0 +1,74 @@+{-# LANGUAGE ScopedTypeVariables #-}++module Network.Socketed.Internal where++import Conduit (+ ConduitM, MonadIO,+ (.|), runConduit, sourceHandle, stdoutC, await, yield, leftover+ )++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (Async, async, cancel, race, wait)+import Control.Exception (AsyncException(..), catch)++import Data.ByteString (ByteString, empty)+import Data.Either.Utils (fromRight)+import qualified Data.Conduit.Binary as CB (lines)++import Language.Haskell.TH (stringE)+import Language.Haskell.TH.Quote (QuasiQuoter(..))++import System.IO (stdin)++data SocketedOptions = SocketedOptions {+ host :: String,+ port :: Int+ }++stdinLines :: MonadIO m => ConduitM a ByteString m ()+stdinLines = sourceHandle stdin .| CB.lines++takeUntil2Empty :: MonadIO m => ConduitM ByteString ByteString m ()+takeUntil2Empty =+ loop+ where+ loop = do+ a <- await+ b <- await+ case (a, b) of+ (Nothing, Nothing) -> loop+ (Just x, Nothing) -> leftover x >> loop+ (Nothing, Just x) -> leftover x >> loop+ (Just x, Just y)+ | x == empty && y == empty -> return ()+ | otherwise -> leftover y >> yield x >> loop++stdinPassthrough :: IO ()+stdinPassthrough = runConduit $ sourceHandle stdin .| stdoutC++withStdinPassthrough :: IO a -> IO a+withStdinPassthrough work = do+ a <- async work+ stdinPassthrough+ wait a++showWSHost :: String -> Int -> String+showWSHost h p = "ws://" ++ h ++ ":" ++ show p++stringQuote :: QuasiQuoter+stringQuote = QuasiQuoter {+ quoteExp = stringE,+ quotePat = undefined,+ quoteType = undefined,+ quoteDec = undefined+}++waitTimeout :: forall a . Async a -> a -> Int -> IO a+waitTimeout task def time =+ let+ left = threadDelay time >> cancel task+ right = wait task+ hdl :: AsyncException -> IO a+ hdl _ = return def+ in+ fmap fromRight (race left right) `catch` hdl
+ Network/Socketed/Template.hs view
@@ -0,0 +1,83 @@+{-# LANGUAGE QuasiQuotes #-}++module Network.Socketed.Template (socketedScript, evalHtml) where++import Data.List (intercalate)++import Network.Socketed.Internal (SocketedOptions(..), showWSHost, stringQuote)++socketedScript :: String -> Int -> String -> String -> String+socketedScript h ldrop hf ef = [stringQuote|+ <script>+ (function(host, linesToDrop, handleMessage, handleMessageError) {+ window.handleMessage = handleMessage;+ window.handleMessageError = handleMessageError;++ var retryCount;++ var connect = function(l) {+ var+ lineDropped = 0,+ socket = new WebSocket(host);++ socket.onopen = function(e) {+ console.log('connected');+ retryCount = 0;+ };++ socket.onclose = function(e) {+ console.log('socket closed: ', e);++ if (++retryCount > 20) {+ window.document.body.textContent = 'disconnected...';+ } else {+ setTimeout(function() {+ console.log('reconnecting...');+ connect(linesToDrop);+ }, 2000);+ }+ };++ socket.onmessage = function(event) {+ if (++lineDropped > l) {+ try {+ window.handleMessage(event);+ } catch(error) {+ window.handleMessageError(event, error);+ }+ } else {+ console.log('drops: ', event.data)+ }+ };+ };++ connect(0);+ })(|]+ ++ intercalate "," ["'" ++ h ++ "'", show ldrop, hf, ef]+ ++ [stringQuote|)</script>|]++wrapHtml :: String -> String+wrapHtml inner = [stringQuote|+ <!DOCTYPE html><html><body>+ |]+ ++ inner+ ++ [stringQuote|+ </body></html>+ |]++evalHtml :: Int -> SocketedOptions -> String+evalHtml r (SocketedOptions h p) = wrapHtml $ socketedScript+ (showWSHost h p)+ r+ [stringQuote|+ function(event) {+ eval(event.data);+ }+ |]+ [stringQuote|+ function(event, error) {+ var c = document.createElement('div');+ c.textContent = 'failed to eval: '+ event.data;+ window.document.body.appendChild(c);+ }+ |]
+ socketed.cabal view
@@ -0,0 +1,66 @@+name: socketed+version: 0.1.0.0++synopsis: simpe tool to serve piped data over http and websocket+description: simpe tool to serve piped data over http and websocket++license: MIT+license-file: LICENSE+author: sen.cenan@gmail.com+maintainer: sen.cenan@gmail.com++category: Web+build-type: Simple+cabal-version: >=1.10++library+ build-depends:+ base >=4.9 && <4.10,+ bytestring >= 0.10.8.1,+ websockets >= 0.10.0.0,+ conduit-combinators >= 1.1.1,+ conduit-extra >= 1.1.15,+ http-types >= 0.9.1,+ warp >= 3.2.11.1,+ text >= 1.2.2.1,+ async >= 2.1.1,+ wai >= 3.2.1.1,+ wai-websockets >= 3.0.1.1,+ stm >= 2.4.4.1,+ stm-chans >= 3.0.0.4,+ stm-conduit >= 3.0.0,+ template-haskell >= 2.11.1.0,+ optparse-applicative >= 0.13.2.0,+ MissingH >= 1.4.0.1+ default-language: Haskell2010+ exposed-modules:+ Network.Socketed,+ Network.Socketed.Template,+ Network.Socketed.Internal++executable socketed+ main-is: Main.hs+ build-depends:+ base >=4.9 && <4.10,+ bytestring >= 0.10.8.1,+ websockets >= 0.10.0.0,+ conduit-combinators >= 1.1.1,+ conduit-extra >= 1.1.15,+ http-types >= 0.9.1,+ warp >= 3.2.11.1,+ text >= 1.2.2.1,+ async >= 2.1.1,+ wai >= 3.2.1.1,+ wai-websockets >= 3.0.1.1,+ stm >= 2.4.4.1,+ stm-chans >= 3.0.0.4,+ stm-conduit >= 3.0.0,+ template-haskell >= 2.11.1.0,+ optparse-applicative >= 0.13.2.0,+ MissingH >= 1.4.0.1+ default-language: Haskell2010+ ghc-options: -threaded+ other-modules:+ Network.Socketed,+ Network.Socketed.Template,+ Network.Socketed.Internal