diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Network/Pusher/WebSockets.hs b/Network/Pusher/WebSockets.hs new file mode 100644 --- /dev/null +++ b/Network/Pusher/WebSockets.hs @@ -0,0 +1,141 @@ +-- | +-- Module : Network.Pusher.WebSockets +-- Copyright : (c) 2016 Michael Walker +-- License : MIT +-- Maintainer : Michael Walker +-- 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 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 diff --git a/Network/Pusher/WebSockets/Channel.hs b/Network/Pusher/WebSockets/Channel.hs new file mode 100644 --- /dev/null +++ b/Network/Pusher/WebSockets/Channel.hs @@ -0,0 +1,131 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | +-- Module : Network.Pusher.WebSockets.Channel +-- Copyright : (c) 2016 Michael Walker +-- License : MIT +-- Maintainer : Michael Walker +-- 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 diff --git a/Network/Pusher/WebSockets/Event.hs b/Network/Pusher/WebSockets/Event.hs new file mode 100644 --- /dev/null +++ b/Network/Pusher/WebSockets/Event.hs @@ -0,0 +1,140 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | +-- Module : Network.Pusher.WebSockets.Event +-- Copyright : (c) 2016 Michael Walker +-- License : MIT +-- Maintainer : Michael Walker +-- 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_)] + ] diff --git a/Network/Pusher/WebSockets/Internal.hs b/Network/Pusher/WebSockets/Internal.hs new file mode 100644 --- /dev/null +++ b/Network/Pusher/WebSockets/Internal.hs @@ -0,0 +1,340 @@ +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE ScopedTypeVariables #-} + +-- | +-- Module : Network.Pusher.WebSockets.Internal +-- Copyright : (c) 2016 Michael Walker +-- License : MIT +-- Maintainer : Michael Walker +-- 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 +-- . +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 + -- . + -- + -- 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) = "<>" + +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) = "<>" + +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 diff --git a/Network/Pusher/WebSockets/Internal/Client.hs b/Network/Pusher/WebSockets/Internal/Client.hs new file mode 100644 --- /dev/null +++ b/Network/Pusher/WebSockets/Internal/Client.hs @@ -0,0 +1,212 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | +-- Module : Network.Pusher.WebSockets.Internal.Client +-- Copyright : (c) 2016 Michael Walker +-- License : MIT +-- Maintainer : Michael Walker +-- 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 diff --git a/Network/Pusher/WebSockets/Internal/Event.hs b/Network/Pusher/WebSockets/Internal/Event.hs new file mode 100644 --- /dev/null +++ b/Network/Pusher/WebSockets/Internal/Event.hs @@ -0,0 +1,133 @@ +{-# LANGUAGE OverloadedStrings #-} + +-- | +-- Module : Network.Pusher.WebSockets.Internal.Event +-- Copyright : (c) 2016 Michael Walker +-- License : MIT +-- Maintainer : Michael Walker +-- 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) diff --git a/Network/Pusher/WebSockets/Util.hs b/Network/Pusher/WebSockets/Util.hs new file mode 100644 --- /dev/null +++ b/Network/Pusher/WebSockets/Util.hs @@ -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 +-- 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" diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/pusher-ws.cabal b/pusher-ws.cabal new file mode 100644 --- /dev/null +++ b/pusher-ws.cabal @@ -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 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