pusher-ws (empty) → 0.1.0.0
raw patch · 10 files changed
+1297/−0 lines, 10 filesdep +aesondep +basedep +bytestringsetup-changed
Dependencies added: aeson, base, bytestring, containers, deepseq, hashable, http-conduit, lens, lens-aeson, network, scientific, stm, text, time, transformers, unordered-containers, websockets, wuss
Files
- LICENSE +20/−0
- Network/Pusher/WebSockets.hs +141/−0
- Network/Pusher/WebSockets/Channel.hs +131/−0
- Network/Pusher/WebSockets/Event.hs +140/−0
- Network/Pusher/WebSockets/Internal.hs +340/−0
- Network/Pusher/WebSockets/Internal/Client.hs +212/−0
- Network/Pusher/WebSockets/Internal/Event.hs +133/−0
- Network/Pusher/WebSockets/Util.hs +93/−0
- Setup.hs +2/−0
- pusher-ws.cabal +85/−0
+ LICENSE view
@@ -0,0 +1,20 @@+Copyright (c) 2016 Michael Walker++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.
+ Network/Pusher/WebSockets.hs view
@@ -0,0 +1,141 @@+-- |+-- Module : Network.Pusher.WebSockets+-- Copyright : (c) 2016 Michael Walker+-- License : MIT+-- Maintainer : Michael Walker <mike@barrucadu.co.uk>+-- Stability : experimental+-- Portability : portable+--+-- Pusher has two APIs: the REST API and the websocket API. The+-- websocket API, which is what this package implements, is used by+-- clients primarily to subscribe to channels and receive events. This+-- library encourages a callback-style approach to Pusher, where the+-- 'pusherWithOptions' function is used to subscribe to some channels+-- and bind some event handlers, and then block until the connection+-- is closed.+--+-- A small example, which simply prints all received events:+--+-- > let key = "your-key"+-- > let channels = ["your", "channels"]+-- >+-- > -- Connect to Pusher with your key, SSL, and the us-east-1 region,+-- > -- and do some stuff.+-- > pusherWithOptions (defaultOptions key) $ do+-- > -- Subscribe to all the channels+-- > mapM_ subscribe channels+-- >+-- > -- Bind an event handler for all events on all channels which+-- > -- prints the received JSON.+-- > bindAll Nothing (liftIO . print)+-- >+-- > -- Wait for user input and then close the connection.+-- > liftIO (void getLine)+-- > disconnectBlocking+--+-- See <https://pusher.com/docs/pusher_protocol> for details of the+-- protocol.+module Network.Pusher.WebSockets+ ( -- * Pusher+ PusherClient+ , PusherClosed(..)+ , AppKey(..)+ , Options(..)+ , Cluster(..)+ , pusherWithOptions+ , defaultOptions++ -- ** Connection+ , ConnectionState(..)+ , connectionState+ , disconnect+ , disconnectBlocking+ , blockUntilDisconnected++ -- * Re-exports+ , module Network.Pusher.WebSockets.Channel+ , module Network.Pusher.WebSockets.Event+ , module Network.Pusher.WebSockets.Util+ ) where++-- 'base' imports+import Control.Concurrent (forkIO)++-- library imports+import Control.Concurrent.STM (atomically, retry)+import Control.Concurrent.STM.TVar (readTVar, writeTVar)+import Control.Monad.IO.Class (liftIO)+import Data.Time.Clock (getCurrentTime)+import Network.WebSockets (runClientWith)+import qualified Network.WebSockets as WS+import Wuss (runSecureClientWith)++-- local imports+import Network.Pusher.WebSockets.Channel+import Network.Pusher.WebSockets.Event+import Network.Pusher.WebSockets.Internal+import Network.Pusher.WebSockets.Internal.Client+import Network.Pusher.WebSockets.Util++-- Haddock doesn't like the import/export shortcut when generating+-- docs.+{-# ANN module "HLint: ignore Use import/export shortcut" #-}++-- | Connect to Pusher.+--+-- This does NOT automatically disconnect from Pusher when the+-- supplied action terminates, so either the action will need to call+-- 'disconnect' or 'disconnectBlocking' as the last thing it does, or+-- one of the event handlers will need to do so eventually.+pusherWithOptions :: Options -> PusherClient a -> IO a+pusherWithOptions opts action+ | encrypted opts = run (runSecureClientWith host port path)+ | otherwise = run (runClientWith host (fromIntegral port) path)++ where+ (host, port, path) = makeURL opts++ -- Run the client+ run withConn = do+ pusher <- defaultPusher opts++ let connOpts = WS.defaultConnectionOptions+ { WS.connectionOnPong = atomically . writeTVar (lastReceived pusher) =<< getCurrentTime }+ let withConnection = withConn connOpts []++ _ <- forkIO (pusherClient pusher withConnection)+ runPusherClient pusher action++-- | Get the connection state.+connectionState :: PusherClient ConnectionState+connectionState = readTVarIO . connState =<< ask++-- | Gracefully close the connection. The connection will remain open+-- and events will continue to be processed until the server accepts+-- the request.+disconnect :: PusherClient ()+disconnect = do+ pusher <- ask+ liftIO (sendCommand pusher Terminate)++-- | Like 'disconnect', but block until the connection is actually+-- closed.+disconnectBlocking :: PusherClient ()+disconnectBlocking = do+ disconnect+ blockUntilDisconnected++-- | Block until the connection is closed (but do not initiate a+-- disconnect).+--+-- This is useful if you run 'pusherWithOptions' in the main thread to+-- prevent the program from terminating until one of your event+-- handlers decides to disconnect.+blockUntilDisconnected :: PusherClient ()+blockUntilDisconnected = do+ pusher <- ask+ liftIO . atomically $ do+ cstate <- readTVar (connState pusher)+ case cstate of+ Disconnected _ -> pure ()+ _ -> retry
+ Network/Pusher/WebSockets/Channel.hs view
@@ -0,0 +1,131 @@+{-# LANGUAGE OverloadedStrings #-}++-- |+-- Module : Network.Pusher.WebSockets.Channel+-- Copyright : (c) 2016 Michael Walker+-- License : MIT+-- Maintainer : Michael Walker <mike@barrucadu.co.uk>+-- Stability : experimental+-- Portability : OverloadedStrings+--+-- Functions for subscribing to and querying channels.+module Network.Pusher.WebSockets.Channel+ ( Channel+ , subscribe+ , unsubscribe+ , members+ , whoami+ ) where++-- 'base' imports+import Data.Monoid ((<>))++-- library imports+import Control.Lens ((&), (%~), ix)+import Control.Monad.IO.Class (MonadIO(..))+import Data.Aeson (Value(..), decode)+import qualified Data.HashMap.Strict as H+import Data.Text (Text, isPrefixOf, pack)+import Data.Text.Encoding (encodeUtf8)+import qualified Network.HTTP.Conduit as W++-- local imports+import Network.Pusher.WebSockets.Event+import Network.Pusher.WebSockets.Internal++-------------------------------------------------------------------------------++-- | Subscribe to a channel. If the channel name begins with+-- \"private-\" or \"presence-\", authorisation is performed+-- automatically.+--+-- This returns immediately. You should wait for the+-- @"pusher:subscription_succeeded"@ event before attempting to use+-- presence channel functions like 'members' and 'whoami'.+--+-- If authorisation fails, this returns @Nothing@.+subscribe :: Text -> PusherClient (Maybe Channel)+subscribe channel = do+ pusher <- ask+ data_ <- getSubscribeData+ case data_ of+ Just (Object o) -> do+ let channelData = Object (H.insert "channel" (String channel) o)+ liftIO (sendCommand pusher (Subscribe handle channelData))++ pure (Just handle)+ _ -> pure Nothing++ where+ getSubscribeData+ | "private-" `isPrefixOf` channel = authorise handle+ | "presence-" `isPrefixOf` channel = authorise handle+ | otherwise = pure (Just (Object H.empty))++ handle = Channel channel++-- | Unsubscribe from a channel.+unsubscribe :: Channel -> PusherClient ()+unsubscribe channel = do+ -- Send the unsubscribe message+ triggerEvent "pusher:unsubscribe" (Just channel) Null++ -- Remove the presence channel+ pusher <- ask+ strictModifyTVarIO (presenceChannels pusher) (H.delete channel)++-- | Return the list of all members in a presence channel.+--+-- If we have unsubscribed from this channel, or it is not a presence+-- channel, returns an empty map.+members :: Channel -> PusherClient (H.HashMap Text Value)+members channel = do+ pusher <- ask+ chan <- H.lookup channel <$> readTVarIO (presenceChannels pusher)+ pure (maybe H.empty snd chan)+ +-- | Return information about the local user in a presence channel.+--+-- If we have unsubscribed from this channel, or it is not a presence+-- channel, returns @Null@.+whoami :: Channel -> PusherClient Value+whoami channel = do+ pusher <- ask++ chan <- H.lookup channel <$> readTVarIO (presenceChannels pusher)+ pure (maybe Null fst chan)++-------------------------------------------------------------------------------++-- | Send a channel authorisation request+authorise :: Channel -> PusherClient (Maybe Value)+authorise (Channel channel) = do+ pusher <- ask+ let authURL = authorisationURL (options pusher)+ let AppKey key = appKey (options pusher)+ sockID <- readTVarIO (socketId pusher)++ case (authURL, sockID) of+ (Just authURL', Just sockID') -> do+ authData <- liftIO (authorise' authURL' sockID')+ pure $ case authData of+ -- If authed, prepend the app key to the "auth" field.+ Just val -> Just (val & ix "auth" %~ prepend (key ++ ":"))+ _ -> Nothing+ _ -> pure Nothing++ where+ -- attempt to authorise against the server.+ authorise' authURL sockID = ignoreAll Nothing $ do+ man <- W.newManager W.tlsManagerSettings+ req <- W.parseUrl authURL+ let req' = W.setQueryString+ [ ("channel_name", Just (encodeUtf8 channel))+ , ("socket_id", Just (encodeUtf8 sockID))+ ] req+ resp <- W.httpLbs req' man+ pure . decode $ W.responseBody resp++ -- prepend a value to a JSON string.+ prepend s (String str) = String (pack s <> str)+ prepend _ val = val
+ Network/Pusher/WebSockets/Event.hs view
@@ -0,0 +1,140 @@+{-# LANGUAGE OverloadedStrings #-}++-- |+-- Module : Network.Pusher.WebSockets.Event+-- Copyright : (c) 2016 Michael Walker+-- License : MIT+-- Maintainer : Michael Walker <mike@barrucadu.co.uk>+-- Stability : experimental+-- Portability : OverloadedStrings+--+-- Functions for creating event handlers and triggering events.+module Network.Pusher.WebSockets.Event+ ( eventType+ , eventChannel++ -- * Event Handlers+ , Binding+ , bind+ , bindAll+ , unbind++ -- * Client Events+ , triggerEvent+ , localEvent+ ) where++-- 'base' imports+import Data.Maybe (fromMaybe)++-- library imports+import Control.Concurrent.STM (atomically, readTVar)+import Control.Lens ((^?), (.~), (&), ix)+import Control.Monad.IO.Class (liftIO)+import Data.Aeson (Value(..), decodeStrict')+import Data.Aeson.Lens (_String)+import qualified Data.HashMap.Strict as H+import Data.Text (Text)+import Data.Text.Encoding (encodeUtf8)++-- local imports+import Network.Pusher.WebSockets.Internal++-------------------------------------------------------------------------------++-- | Get the value of the \"event\" field.+--+-- If not present (which should never happen!), returns the empty+-- string.+eventType :: Value -> Text+eventType event = fromMaybe "" (event ^? ix "event" . _String)++-- | Get the value of the \"channel\" field.+--+-- This will be @Nothing@ if the event was broadcast to all clients,+-- with no channel restriction.+eventChannel :: Value -> Maybe Channel+eventChannel event = fmap Channel (event ^? ix "channel" . _String)++-------------------------------------------------------------------------------++-- | Bind an event handler to an event type, optionally restricted to a+-- channel.+--+-- Attempts to decode the \"data\" field of the event as stringified+-- JSON; if that fails, it is left as a string.+--+-- If multiple handlers match a received event, all will be+-- executed. The order is unspecified, and may not be consistent.+bind :: Text+ -- ^ Event name.+ -> Maybe Channel+ -- ^ Channel name: If @Nothing@, all events of that name are+ -- handled.+ -> (Value -> PusherClient ())+ -- ^ Event handler.+ -> PusherClient Binding+bind = bindGeneric . Just++-- | Variant of 'bind' which binds to all events in the given channel;+-- or all events if no channel.+bindAll :: Maybe Channel -> (Value -> PusherClient ()) -> PusherClient Binding+bindAll = bindGeneric Nothing++-- | Internal: register a new event handler.+bindGeneric :: Maybe Text -> Maybe Channel -> (Value -> PusherClient ())+ -> PusherClient Binding+bindGeneric event channel handler = do+ pusher <- ask+ liftIO . atomically $ do+ b@(Binding i) <- readTVar (nextBinding pusher)+ let b' = Binding (i+1)+ strictModifyTVar (nextBinding pusher) (const b')+ let h = Handler event channel wrappedHandler+ strictModifyTVar (eventHandlers pusher) (H.insert b h)+ pure b++ where+ -- Before invoking the handler, have a stab at decoding the data+ -- field.+ wrappedHandler ev@(Object o) = handler $+ case H.lookup "data" o >>= attemptDecode of+ Just decoded -> ev & ix "data" .~ decoded+ Nothing -> ev+ wrappedHandler ev = handler ev++ -- Attempt to interpret as stringified JSON.+ attemptDecode (String s) = decodeStrict' (encodeUtf8 s)+ attemptDecode _ = Nothing++-- | Remove a binding+unbind :: Binding -> PusherClient ()+unbind binding = do+ pusher <- ask+ strictModifyTVarIO (eventHandlers pusher) (H.delete binding)++-------------------------------------------------------------------------------++-- | Send an event with some JSON data. This does not trigger local+-- event handlers.+triggerEvent :: Text -> Maybe Channel -> Value -> PusherClient ()+triggerEvent = sendMessage SendMessage++-- | Trigger local event handlers, but do not send the event over the+-- network.+localEvent :: Text -> Maybe Channel -> Value -> PusherClient ()+localEvent = sendMessage SendLocalMessage++-- | Helper function for 'triggerEvent' and 'localEvent'+sendMessage :: (Value -> PusherCommand)+ -> Text -> Maybe Channel -> Value -> PusherClient ()+sendMessage cmd event channel data_ = do+ pusher <- ask+ liftIO (sendCommand pusher (cmd json))++ where+ json = Object . H.fromList $ concat+ [ [("event", String event)]+ , [("channel", String chan) | Just (Channel chan) <- [channel]]+ , [("data", data_)]+ ]
+ Network/Pusher/WebSockets/Internal.hs view
@@ -0,0 +1,340 @@+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- |+-- Module : Network.Pusher.WebSockets.Internal+-- Copyright : (c) 2016 Michael Walker+-- License : MIT+-- Maintainer : Michael Walker <mike@barrucadu.co.uk>+-- Stability : experimental+-- Portability : GeneralizedNewtypeDeriving, ScopedTypeVariables+--+-- Internal types and functions. This is NOT considered to form part+-- of the public API of this library.+module Network.Pusher.WebSockets.Internal where++-- 'base' imports+import Control.Concurrent (ThreadId)+import Control.Exception (IOException, Exception, SomeException, catch, toException)+import qualified Control.Exception as E+import Data.String (IsString(..))+import Data.Word (Word16)++-- library imports+import Control.Concurrent.STM (STM, TVar, atomically, newTVar, modifyTVar')+import Control.Concurrent.STM.TQueue+import qualified Control.Concurrent.STM as STM+import Control.DeepSeq (NFData(..), force)+import Control.Monad.IO.Class (MonadIO(..))+import Control.Monad.Trans.Reader (ReaderT, runReaderT)+import qualified Control.Monad.Trans.Reader as R+import Data.Aeson (Value(..))+import Data.Hashable (Hashable(..))+import qualified Data.HashMap.Strict as H+import qualified Data.Set as S+import Data.Text (Text, unpack)+import Data.Time.Clock (UTCTime, getCurrentTime)+import Network.Socket (HostName, PortNumber)+import Network.WebSockets (ConnectionException, HandshakeException)++-------------------------------------------------------------------------------++-- | A value of type @PusherClient a@ is a computation with access to+-- a connection to Pusher which, when executed, may perform+-- Pusher-specific actions such as subscribing to channels and+-- receiving events, as well as arbitrary I/O.+newtype PusherClient a = PusherClient (ReaderT Pusher IO a)+ deriving (Functor, Applicative, Monad, MonadIO)++-- | Run a 'PusherClient'.+runPusherClient :: Pusher -> PusherClient a -> IO a+runPusherClient pusher (PusherClient action) = runReaderT action pusher++-- | Pusher connection handle.+--+-- If this is used after disconnecting, an exception will be thrown.+data Pusher = Pusher+ { commandQueue :: TQueue PusherCommand+ -- ^ Queue to send commands to the client thread.+ , connState :: TVar ConnectionState+ -- ^ The state of the connection.+ , options :: Options+ -- ^ Connection options+ , idleTimer :: TVar (Maybe Int)+ -- ^ Inactivity timeout before a ping should be sent. Set by Pusher+ -- on connect.+ , lastReceived :: TVar UTCTime+ -- ^ Time of receipt of last message.+ , socketId :: TVar (Maybe Text)+ -- ^ Identifier of the socket. Set by Pusher on connect.+ , threadStore :: TVar (S.Set ThreadId)+ -- ^ Currently live threads.+ , eventHandlers :: TVar (H.HashMap Binding Handler)+ -- ^ Event handlers.+ , nextBinding :: TVar Binding+ -- ^ Next free binding.+ , allChannels :: TVar (S.Set Channel)+ -- ^ All subscribed channels.+ , presenceChannels :: TVar (H.HashMap Channel (Value, H.HashMap Text Value))+ -- ^ Connected presence channels+ }++-- | A command to the Pusher thread.+data PusherCommand+ = SendMessage Value+ -- ^ Send a message over the network, not triggering event handlers.+ | SendLocalMessage Value+ -- ^ Do not send a message over the network, trigger event handlers.+ | Subscribe Channel Value+ -- ^ Send a channel subscription message and add to the+ -- 'allChannels' set.+ | Terminate+ -- ^ Gracefully close the connection.+ deriving (Eq, Show)++-- | An exception thrown to kill the client.+data TerminatePusher = TerminatePusher (Maybe Word16)+ deriving (Eq, Ord, Read, Show)++instance Exception TerminatePusher++-- | Thrown if attempting to communicate with Pusher after the+-- connection has been closed.+--+-- If the server closed the connection, the error code is+-- included. See the 4000-4099 error codes on+-- <https://pusher.com/docs/pusher_protocol>.+data PusherClosed = PusherClosed (Maybe Word16)+ deriving (Eq, Ord, Read, Show)++instance Exception PusherClosed++-- | The state of the connection. Events are sent when the state is+-- changed.+data ConnectionState+ = Initialized+ -- ^ Initial state. No event is emitted.+ | Connecting+ -- ^ Trying to connect. This state will also be entered when trying+ -- to reconnect after a connection failure.+ --+ -- Emits the @"connecting"@ event.+ | Connected+ -- ^ The connection is established and authenticated with your+ -- app.+ --+ -- Emits the @"connected"@ event.+ | Unavailable+ -- ^ The connection is temporarily unavailable. The network+ -- connection is down, the server is down, or something is blocking+ -- the connection.+ --+ -- Emits the @"unavailable"@ event and then enters the @Connecting@+ -- state again.+ | Disconnected (Maybe Word16)+ -- ^ The connection has been closed by the client, or the server+ -- indicated an error which cannot be resolved by reconnecting with+ -- the same settings.+ --+ -- If the server closed the connection, the error code is+ -- included. See the 4000-4099 error codes on+ -- <https://pusher.com/docs/pusher_protocol>.+ --+ -- Emits the @"disconnected"@ event and then kills all forked+ -- threads.+ deriving (Eq, Ord, Read, Show)++-- | State for a brand new connection.+defaultPusher :: Options -> IO Pusher+defaultPusher opts = do+ now <- getCurrentTime+ atomically $ do+ defCommQueue <- newTQueue+ defConnState <- newTVar Initialized+ defIdleTimer <- newTVar Nothing+ defLastReceived <- newTVar now+ defSocketId <- newTVar Nothing+ defThreadStore <- newTVar S.empty+ defEHandlers <- newTVar H.empty+ defBinding <- newTVar (Binding 0)+ defAChannels <- newTVar S.empty+ defPChannels <- newTVar H.empty++ pure Pusher+ { commandQueue = defCommQueue+ , connState = defConnState+ , options = opts+ , idleTimer = defIdleTimer+ , lastReceived = defLastReceived+ , socketId = defSocketId+ , threadStore = defThreadStore+ , eventHandlers = defEHandlers+ , nextBinding = defBinding+ , allChannels = defAChannels+ , presenceChannels = defPChannels+ }++-- | Send a command to the queue. Throw a 'PusherClosed' exception if+-- the connection has been disconnected.+sendCommand :: Pusher -> PusherCommand -> IO ()+sendCommand pusher cmd = do+ cstate <- readTVarIO (connState pusher)+ case cstate of+ Disconnected ccode -> E.throwIO (PusherClosed ccode)+ _ -> atomically (writeTQueue (commandQueue pusher) cmd)++-------------------------------------------------------------------------------++data Options = Options+ { appKey :: AppKey+ -- ^ The application key.++ , encrypted :: Bool+ -- ^ If the connection should be made over an encrypted+ -- connection. Defaults to @True@.++ , authorisationURL :: Maybe String+ -- ^ The URL which will return the authentication signature needed+ -- for private and presence channels. If not given, private and+ -- presence channels cannot be used. Defaults to @Nothing@.++ , cluster :: Cluster+ -- ^ Allows connecting to a different cluster by setting up correct+ -- hostnames for the connection. This parameter is mandatory when+ -- the app is created in a different cluster to the default+ -- us-east-1. Defaults to @MT1@.++ , pusherURL :: Maybe (HostName, PortNumber, String)+ -- ^ The host, port, and path to use instead of the standard Pusher+ -- servers. If set, the cluster is ignored. Defaults to @Nothing@.+ } deriving (Eq, Ord, Show)++instance NFData Options where+ rnf o = rnf ( appKey o+ , encrypted o+ , authorisationURL o+ , cluster o+ , mangle (pusherURL o)+ )+ where+ mangle Nothing = Nothing+ mangle (Just (h, p, s)) = p `seq` Just (h, s)++-- | Clusters correspond to geographical regions where apps can be+-- assigned to.+data Cluster+ = MT1 -- ^ The us-east-1 cluster.+ | EU -- ^ The eu-west-1 cluster.+ | AP1 -- ^ The ap-southeast-1 cluster.+ deriving (Eq, Ord, Bounded, Enum, Read, Show)++instance NFData Cluster where+ rnf c = c `seq` ()++-- | Your application's API key.+newtype AppKey = AppKey String+ deriving (Eq, Ord, Show, Read)++instance IsString AppKey where+ fromString = AppKey++instance NFData AppKey where+ rnf (AppKey k) = rnf k++-- | See 'Options' field documentation for what is set here.+defaultOptions :: AppKey -> Options+defaultOptions key = Options+ { appKey = key+ , encrypted = True+ , authorisationURL = Nothing+ , cluster = MT1+ , pusherURL = Nothing+ }++-------------------------------------------------------------------------------++-- | Event handlers: event name -> channel name -> handler.+data Handler = Handler (Maybe Text) (Maybe Channel) (Value -> PusherClient ())++-- Cheats a bit.+instance NFData Handler where+ rnf (Handler e c _) = rnf (e, c)++-------------------------------------------------------------------------------++-- | Channel handle: a witness that we joined a channel, and is used+-- to subscribe to events.+--+-- If this is used when unsubscribed from a channel, nothing will+-- happen.+newtype Channel = Channel { unChannel :: Text }+ deriving (Eq, Ord)++instance NFData Channel where+ rnf (Channel c) = rnf c++instance Show Channel where+ show (Channel c) = "<<channel " ++ unpack c ++ ">>"++instance Hashable Channel where+ hashWithSalt salt (Channel c) = hashWithSalt salt c++-------------------------------------------------------------------------------++-- | Event binding handle: a witness that we bound an event handler, and is+-- used to unbind it.+--+-- If this is used after unbinding, nothing will happen.+newtype Binding = Binding { unBinding :: Int }+ deriving (Eq, Ord)++instance NFData Binding where+ rnf (Binding b) = rnf b++instance Show Binding where+ show (Binding b) = "<<binding " ++ show b ++ ">>"++instance Hashable Binding where+ hashWithSalt salt (Binding b) = hashWithSalt salt b++-------------------------------------------------------------------------------++-- | Get the current state.+ask :: PusherClient Pusher+ask = PusherClient R.ask++-- | Modify a @TVar@ strictly.+strictModifyTVar :: NFData a => TVar a -> (a -> a) -> STM ()+strictModifyTVar tvar = modifyTVar' tvar . force++-- | Modify a @TVar@ strictly in any @MonadIO@.+strictModifyTVarIO :: (MonadIO m, NFData a) => TVar a -> (a -> a) -> m ()+strictModifyTVarIO tvar = liftIO . atomically . strictModifyTVar tvar++-- | Read a @TVar@ inside any @MonadIO@.+readTVarIO :: MonadIO m => TVar a -> m a+readTVarIO = liftIO . STM.readTVarIO++-------------------------------------------------------------------------------++-- | Ignore all exceptions by supplying a default value.+ignoreAll :: a -> IO a -> IO a+ignoreAll fallback act = catchAll act (const (pure fallback))++-- | Run an action, starting again on connection and handshake+-- exception.+reconnecting :: IO a -> IO () -> IO a+reconnecting act prere = loop where+ loop = catchNetException act (const (prere >> loop))++-- | Catch all network exceptions.+catchNetException :: forall a. IO a -> (SomeException -> IO a) -> IO a+catchNetException act handler = E.catches act handlers where+ handlers = [ E.Handler (handler . toException :: IOException -> IO a)+ , E.Handler (handler . toException :: HandshakeException -> IO a)+ , E.Handler (handler . toException :: ConnectionException -> IO a)+ ]++-- | Catch all exceptions.+catchAll :: IO a -> (SomeException -> IO a) -> IO a+catchAll = catch
+ Network/Pusher/WebSockets/Internal/Client.hs view
@@ -0,0 +1,212 @@+{-# LANGUAGE OverloadedStrings #-}++-- |+-- Module : Network.Pusher.WebSockets.Internal.Client+-- Copyright : (c) 2016 Michael Walker+-- License : MIT+-- Maintainer : Michael Walker <mike@barrucadu.co.uk>+-- Stability : experimental+-- Portability : OverloadedStrings+--+-- Pusher network client. This is NOT considered to form part of the+-- public API of this library.+module Network.Pusher.WebSockets.Internal.Client where++-- 'base' imports+import Control.Concurrent (forkIO, killThread, threadDelay)+import Control.Exception (fromException, throwIO)+import Control.Monad (forever)+import Data.Maybe (isJust)++-- library imports+import Control.Concurrent.STM (atomically, check, retry)+import Control.Concurrent.STM.TQueue (tryReadTQueue)+import Control.Concurrent.STM.TVar (TVar, newTVarIO, readTVar, writeTVar)+import Control.Monad.IO.Class (liftIO)+import Data.Aeson (Value(..), encode)+import qualified Data.HashMap.Strict as H+import qualified Data.Set as S+import Data.Text (Text, pack)+import Data.Time.Clock (getCurrentTime, diffUTCTime)+import Data.Word (Word16)+import Network.WebSockets (Connection, sendClose)+import qualified Network.WebSockets as WS++-- local imports+import Network.Pusher.WebSockets.Channel+import Network.Pusher.WebSockets.Event+import Network.Pusher.WebSockets.Internal+import Network.Pusher.WebSockets.Internal.Event+import Network.Pusher.WebSockets.Util++-------------------------------------------------------------------------------+-- Pusher client++-- | Client thread: connect to Pusher and process commands,+-- reconnecting automatically, until finally told to terminate.+--+-- Does not automatically fork.+pusherClient :: Pusher -> ((Connection -> IO ()) -> IO ()) -> IO ()+pusherClient pusher withConnection = do+ -- Bind default handlers+ runPusherClient pusher $+ mapM_ (\(e, h) -> bind e Nothing h) defaultHandlers++ -- Run client+ catchAll+ (reconnecting+ (changeConnectionState pusher Connecting >>+ withConnection (client pusher))+ (changeConnectionState pusher Unavailable >>+ threadDelay (1 * 1000 * 1000)))+ (\e -> case fromException e of+ Just (TerminatePusher closeCode) ->+ changeConnectionState pusher (Disconnected closeCode)+ Nothing ->+ changeConnectionState pusher (Disconnected Nothing))++ -- Kill forked threads+ readTVarIO (threadStore pusher) >>= mapM_ killThread++-- | Fork off event handling and pinging threads, subscribe to+-- channels, and loop processing commands until terminated.+client :: Pusher -> Connection -> IO ()+client pusher conn = flip catchAll handleExc $ do+ -- Fork off an event handling thread+ closevar <- newTVarIO Nothing+ _ <- forkIO (handleThread pusher conn closevar)++ -- Wait for the pusher:connection_established event.+ liftIO . atomically $+ check . isJust =<< readTVar (idleTimer pusher)+ changeConnectionState pusher Connected++ -- This will do more pinging than necessary, but it's far simpler+ -- than keeping track of the actual inactivity, and ensures that+ -- enough pings are sent.+ _ <- forkIO (pingThread pusher conn closevar)++ -- Subscribe to channels+ channels <- liftIO . atomically $ do+ writeTVar (presenceChannels pusher) H.empty+ readTVar (allChannels pusher)+ runPusherClient pusher $+ mapM_ (subscribe . unChannel) channels++ -- Handle commands+ forever $+ handleCommandOrClose pusher conn =<< awaitCommandOrClose pusher closevar++ where+ -- Mark the connection as closed by clearing the idle timer and+ -- socket ID and rethrow the exception.+ handleExc e = do+ strictModifyTVarIO (idleTimer pusher) (const Nothing)+ strictModifyTVarIO (socketId pusher) (const Nothing)+ throwIO e++-- | Wait for a command or close signal.+awaitCommandOrClose :: Pusher+ -> TVar (Maybe Word16)+ -> IO (Either Word16 PusherCommand)+awaitCommandOrClose pusher closevar = atomically $ do+ cmd <- tryReadTQueue (commandQueue pusher)+ ccode <- readTVar closevar+ case (cmd, ccode) of+ (Just cmd', _) -> pure (Right cmd')+ (Nothing, Just ccode') -> pure (Left ccode')+ (Nothing, Nothing) -> retry++-- | Handle a command or close signal. Throws an exception on+-- disconnect: 'TerminatePusher' if the connection should not be+-- re-established, and 'WS.ConnectionClosed' if it should be.+handleCommandOrClose :: Pusher+ -> Connection+ -> Either Word16 PusherCommand+ -> IO ()+handleCommandOrClose pusher conn (Right pusherCommand) =+ handleCommand pusher conn pusherCommand+handleCommandOrClose _ _ (Left closeCode) =+ throwCloseException closeCode++-- | Handle a command.+handleCommand :: Pusher -> Connection -> PusherCommand -> IO ()+handleCommand pusher conn pusherCommand = case pusherCommand of+ SendMessage json -> sendJSON json+ SendLocalMessage json -> handleEvent pusher (Right json)+ Subscribe handle channelData -> do+ sendJSON . Object $ H.fromList+ [ ("event", String "pusher:subscribe")+ , ("data", channelData)+ ]+ strictModifyTVarIO (allChannels pusher) (S.insert handle)+ Terminate -> sendClose conn ("goodbye" :: Text)+ where+ -- Send some JSON down the channel.+ sendJSON = WS.sendDataMessage conn . WS.Text . encode++-- | Throw the appropriate exception for a close code.+throwCloseException :: Word16 -> IO a+throwCloseException closeCode+ -- Graceful termination+ | closeCode < 4000 =+ throwIO $ TerminatePusher Nothing+ -- Server specified not to reconnect+ | closeCode >= 4000 && closeCode < 4100 =+ throwIO . TerminatePusher $ Just closeCode+ -- Reconnect+ | otherwise =+ throwIO WS.ConnectionClosed++-- | Send a ping every time the timeout elapses. If the connection+-- closes the 'reconnectImmediately' close code is written to the+-- 'TVar'.+pingThread :: Pusher -> Connection -> TVar (Maybe Word16) -> IO ()+pingThread pusher conn closevar = do+ timeout <- liftIO . atomically $+ maybe retry pure =<< readTVar (idleTimer pusher)+ pinger timeout 0++ where+ pinger :: Int -> Integer -> IO ()+ pinger timeout i = do+ -- Wait for the timeout to elapse+ threadDelay (timeout * 1000 * 1000)+ -- Send a ping+ WS.sendPing conn (pack $ show i)+ -- Check the time of receipt of the last message: if it's longer+ -- ago than the timeout signal disconnection. Otherwise loop.+ now <- getCurrentTime+ lastMsg <- readTVarIO (lastReceived pusher)+ if now `diffUTCTime` lastMsg > fromIntegral timeout+ then atomically (writeTVar closevar reconnectImmediately)+ else pinger timeout (i + 1)++-- | Receive and handle events until the connection is closed, at+-- which point the close code is written to the provided 'TVar'.+handleThread :: Pusher -> Connection -> TVar (Maybe Word16) -> IO ()+handleThread pusher conn closevar = handler `catchAll` finaliser+ where+ handler = forever $ do+ msg <- awaitEvent conn+ atomically . writeTVar (lastReceived pusher) =<< getCurrentTime+ handleEvent pusher msg++ finaliser e = atomically . writeTVar closevar $ case fromException e of+ Just (WS.CloseRequest ccode _) -> Just ccode+ _ -> reconnectImmediately++-- | @Just 4200@ = generic reconnect immediately+reconnectImmediately :: Maybe Word16+reconnectImmediately = Just 4200++-- | Set the connection state and send a state change event if+-- necessary.+changeConnectionState :: Pusher -> ConnectionState -> IO ()+changeConnectionState pusher connst = do+ ev <- atomically $ do+ oldState <- readTVar (connState pusher)+ writeTVar (connState pusher) connst+ pure $ (Object . H.singleton "event" . String) <$>+ (if oldState == connst then Nothing else Just (connectionEvent connst))+ maybe (pure ()) (handleEvent pusher . Right) ev
+ Network/Pusher/WebSockets/Internal/Event.hs view
@@ -0,0 +1,133 @@+{-# LANGUAGE OverloadedStrings #-}++-- |+-- Module : Network.Pusher.WebSockets.Internal.Event+-- Copyright : (c) 2016 Michael Walker+-- License : MIT+-- Maintainer : Michael Walker <mike@barrucadu.co.uk>+-- Stability : experimental+-- Portability : OverloadedStrings+--+-- Event handling. This is NOT considered to form part of the public+-- API of this library.+module Network.Pusher.WebSockets.Internal.Event where++-- 'base' imports+import Control.Arrow (second)+import Data.Maybe (isNothing)++-- library imports+import Control.Lens ((&), (^?), (.~), ix)+import Control.Monad.IO.Class (liftIO)+import Data.Aeson (Value(..), decode')+import Data.Aeson.Lens (_Integral, _Object, _String, _Value)+import Data.ByteString.Lazy (ByteString)+import qualified Data.HashMap.Strict as H+import Data.Text (Text)+import Network.WebSockets (Connection, DataMessage(..), receiveDataMessage)++-- local imports+import Network.Pusher.WebSockets.Channel+import Network.Pusher.WebSockets.Event+import Network.Pusher.WebSockets.Internal+import Network.Pusher.WebSockets.Util++-------------------------------------------------------------------------------+-- Handler dispatch++-- | Block and wait for an event.+awaitEvent :: Connection -> IO (Either ByteString Value)+awaitEvent = fmap decode . receiveDataMessage where+ decode (Text bs) = maybe (Left bs) Right (decode' bs)+ decode (Binary bs) = Left bs++-- | Launch all event handlers which are bound to the current event.+handleEvent :: Pusher -> Either ByteString Value -> IO ()+handleEvent pusher (Right event) = do+ let match (Handler e c _) = (isNothing e || e == Just (eventType event)) &&+ (isNothing c || c == eventChannel event)++ handlers <- filter match . H.elems <$> readTVarIO (eventHandlers pusher)+ runPusherClient pusher $+ mapM_ (fork . (\(Handler _ _ h) -> h event)) handlers+-- Discard events which couldn't be decoded.+handleEvent _ _ = pure ()++-------------------------------------------------------------------------------+-- Default handlers++-- | Default event handlers+defaultHandlers :: [(Text, Value -> PusherClient ())]+defaultHandlers =+ [ ("pusher:ping", pingHandler)+ , ("pusher:connection_established", establishConnection)+ , ("pusher_internal:subscription_succeeded", addChannel)+ , ("pusher_internal:member_added", addPresenceMember)+ , ("pusher_internal:member_removed", rmPresenceMember)+ ]++-- | Immediately send a pusher:pong+pingHandler :: Value -> PusherClient ()+pingHandler _ = triggerEvent "pusher:pong" Nothing (Object H.empty)++-- | Record the activity timeout and socket ID.+establishConnection :: Value -> PusherClient ()+establishConnection event = do+ let socketidmay = event ^? ix "data" . ix "socket_id" . _String+ let timeoutmay = event ^? ix "data" . ix "activity_timeout" . _Integral++ case (,) <$> socketidmay <*> timeoutmay of+ Just (socketid, timeout) -> do+ pusher <- ask+ strictModifyTVarIO (idleTimer pusher) (const (Just timeout))+ strictModifyTVarIO (socketId pusher) (const (Just socketid))+ Nothing -> pure ()++-- | Save the list of users (if there is one) and send the internal+-- "pusher:subscription_succeeded" event.+addChannel :: Value -> PusherClient ()+addChannel event = do+ let channelmay = eventChannel event+ let usersmay = event ^? ix "data" . ix "hash" . _Object++ case channelmay of+ Just channel -> do+ pusher <- ask+ maybe (pure ()) (mapUsers channel . const) usersmay+ let json = event & ix "event" .~ "pusher:subscription_succeeded"+ liftIO $ handleEvent pusher (Right json)+ Nothing -> pure ()++-- | Record a presence channel user.+addPresenceMember :: Value -> PusherClient ()+addPresenceMember event = do+ let channelmay = eventChannel event+ let uidmay = event ^? ix "data" . ix "user_id" . _String+ let infomay = event ^? ix "data" . ix "user_info" . _Value++ case (,,) <$> channelmay <*> uidmay <*> infomay of+ Just (channel, uid, info) ->+ mapUsers channel (H.insert uid info)+ Nothing -> pure ()++-- | Remove a presence channel user.+rmPresenceMember :: Value -> PusherClient ()+rmPresenceMember event = do+ let channelmay = eventChannel event+ let uidmay = event ^? ix "data" . ix "user_id" . _String++ case (,) <$> channelmay <*> uidmay of+ Just (channel, uid) ->+ mapUsers channel (H.delete uid)+ Nothing -> pure ()++-------------------------------------------------------------------------------+-- Utilities++-- | Apply a function to the users list of a presence channel+mapUsers :: Channel+ -> (H.HashMap Text Value -> H.HashMap Text Value)+ -> PusherClient ()+mapUsers channel f = do+ pusher <- ask+ strictModifyTVarIO (presenceChannels pusher) (H.adjust (second f) channel)
+ Network/Pusher/WebSockets/Util.hs view
@@ -0,0 +1,93 @@+{-# LANGUAGE OverloadedStrings #-}+{-# OPTIONS -fno-warn-warnings-deprecations #-}++-- |+-- Module : Network.Pusher.WebSockets.Util+-- Copyright : (c) 2016 Michael Walker+-- License : MIT+-- Maintainer : Michael Walker <mike@barrucadu.co.uk>+-- Stability : experimental+-- Portability : OverloadedStrings+--+-- General utility functions.+module Network.Pusher.WebSockets.Util where++-- 'base' imports+import Control.Concurrent (ThreadId, forkIO, myThreadId)+import Control.Exception (bracket_)+import Data.Version (Version(..), showVersion)++-- library imports+import Control.Monad.Trans.Reader (ReaderT(..))+import qualified Data.Set as S+import Data.Text (Text, unpack)+import Network.Socket (HostName, PortNumber)++-- local imports+import Network.Pusher.WebSockets.Internal+import Paths_pusher_ws (version)++-- | Fork a thread which will be killed when the connection is closed.+fork :: PusherClient () -> PusherClient ThreadId+fork (PusherClient (ReaderT action)) = PusherClient $ ReaderT (forkIO . run)+ where+ run s = bracket_ setup teardown (action s) where+ -- Add the thread ID to the list+ setup = do+ tid <- myThreadId+ strictModifyTVarIO (threadStore s) (S.insert tid)++ -- Remove the thread ID from the list+ teardown = do+ tid <- myThreadId+ strictModifyTVarIO (threadStore s) (S.delete tid)++-- | The hostname, port, and path (including querystring) to connect+-- to.+makeURL :: Options -> (HostName, PortNumber, String)+makeURL opts = case pusherURL opts of+ Just (host, port, path) -> (host, port, path ++ queryString)+ Nothing -> (defaultHost, defaultPort, defaultPath)++ where+ defaultHost+ -- The primary cluster has a different domain to all the others+ | cluster opts == MT1 = "ws.pusherapp.com"+ | otherwise = "ws-" ++ theCluster ++ ".pusher.com"++ theCluster = unpack . clusterName $ cluster opts++ defaultPort+ | encrypted opts = 443+ | otherwise = 80++ defaultPath = case appKey opts of+ AppKey k -> "/app/" ++ k ++ queryString++ queryString = "?client=haskell-pusher-ws&protocol=7&version="+ ++ showVersion semver++-- | Three-component semver of the library. This corresponds to the+-- first three components of the Haskell version, as the Pusher+-- servers don't like a four-component version number.+semver :: Version+semver = Version+ { versionBranch = take 3 (versionBranch version)+ , versionTags = []+ }++-- | The region name of a cluster.+clusterName :: Cluster -> Text+clusterName MT1 = "us-east-1"+clusterName EU = "eu-west-1"+clusterName AP1 = "ap-southeast-1"++-- | The event name corresponding to a connection state. 'Initialized'+-- has the "initialized" event for consistency, but this event is+-- never emitted.+connectionEvent :: ConnectionState -> Text+connectionEvent Initialized = "initialized"+connectionEvent Connecting = "connecting"+connectionEvent Connected = "connected"+connectionEvent Unavailable = "unavailable"+connectionEvent (Disconnected _) = "disconnected"
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ pusher-ws.cabal view
@@ -0,0 +1,85 @@+-- Initial pusher-ws.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/++name: pusher-ws+version: 0.1.0.0+synopsis: Implementation of the Pusher WebSocket protocol++description:+ An implementation of the Pusher WebSocket (client) protocol in+ Haskell.+ .+ Current features:+ .+ * @ws://@ and @wss://@ protocols.+ * Clusters.+ * Subscribing to channels.+ * Unsubscribing from channels.+ * Authorisation for private and presence channels.+ * Binding event handlers.+ * Unbinding event handlers+ * Sending client events.+ * Threads which automatically get cleaned up on connection close.+ * Automatic reconnection (and channel resubscription).+ * Connection state events.+ * Pusher close codes.+ .+ Missing features:+ .+ * "connecting_in" events.+ .+ See the <https://github.com/barrucadu/pusher-ws README> for more+ details.+++homepage: https://github.com/barrucadu/pusher-ws+license: MIT+license-file: LICENSE+author: Michael Walker+maintainer: mike@barrucadu.co.uk+-- copyright: +category: Network+build-type: Simple+-- extra-source-files: +cabal-version: >=1.10++source-repository head+ type: git+ location: https://github.com/barrucadu/pusher-ws.git++source-repository this+ type: git+ location: https://github.com/barrucadu/pusher-ws.git+ tag: pusher-ws-0.1.0.0++library+ exposed-modules: Network.Pusher.WebSockets+ , Network.Pusher.WebSockets.Channel+ , Network.Pusher.WebSockets.Event+ , Network.Pusher.WebSockets.Util+ other-modules: Network.Pusher.WebSockets.Internal+ , Network.Pusher.WebSockets.Internal.Client+ , Network.Pusher.WebSockets.Internal.Event+ , Paths_pusher_ws+ -- other-extensions: + build-depends: base >=4.8 && <4.9+ , aeson+ , bytestring+ , containers+ , deepseq+ , hashable+ , http-conduit+ , lens+ , lens-aeson+ , network+ , scientific+ , stm+ , text+ , time+ , transformers+ , unordered-containers+ , websockets+ , wuss+ -- hs-source-dirs: + default-language: Haskell2010+ ghc-options: -Wall