diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -70,7 +70,8 @@ send :: ApnOptions -> IO () send o = do - let mkSession = newSession (keypath o) (certpath o) (capath o) (sandbox o) 10 (B8.pack $ topic o) + let mkSession = + newSession (keypath o) (certpath o) (capath o) (sandbox o) 10 1 (B8.pack $ topic o) session <- mkSession if interactive o then diff --git a/changelog.md b/changelog.md --- a/changelog.md +++ b/changelog.md @@ -1,3 +1,14 @@ +0.2.0.1 +======= + +- Use Data.Pool + +0.2.0.0 +======= + +- Threading bugfix +- Migrate to http2-client-0.9.0.0 and lts-14 + 0.1.1.1 ======= diff --git a/push-notify-apn.cabal b/push-notify-apn.cabal --- a/push-notify-apn.cabal +++ b/push-notify-apn.cabal @@ -1,5 +1,5 @@ name: push-notify-apn -version: 0.2.0.0 +version: 0.2.0.1 synopsis: Send push notifications to mobile iOS devices description: push-notify-apn is a library and command line utility that can be used to send @@ -12,7 +12,7 @@ license-file: LICENSE author: Hans-Christian Esperer maintainer: hc@hcesperer.org -copyright: 2017-2018 Memrange UG +copyright: 2017-2020 Memrange UG category: Network, Cloud, Mobile build-type: Simple extra-source-files: README.md @@ -31,7 +31,10 @@ , data-default , http2 , http2-client + , lifted-base + , mtl , random + , resource-pool , semigroups , text , time diff --git a/src/Network/PushNotify/APN.hs b/src/Network/PushNotify/APN.hs --- a/src/Network/PushNotify/APN.hs +++ b/src/Network/PushNotify/APN.hs @@ -10,6 +10,7 @@ {-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TupleSections #-} module Network.PushNotify.APN @@ -48,8 +49,9 @@ import Control.Concurrent import Control.Concurrent.QSem -import Control.Exception +import Control.Exception.Lifted (Exception, try, bracket_, throw, throwIO) import Control.Monad +import Control.Monad.Except import Data.Aeson import Data.Aeson.Types import Data.ByteString (ByteString) @@ -60,8 +62,10 @@ import Data.IORef import Data.Map.Strict (Map) import Data.Maybe +import Data.Pool import Data.Semigroup ((<>)) import Data.Text (Text) +import Data.Time.Clock import Data.Time.Clock.POSIX import Data.Typeable (Typeable) import Data.X509 @@ -91,10 +95,9 @@ -- | A session that manages connections to Apple's push notification service data ApnSession = ApnSession - { apnSessionPool :: !(IORef [ApnConnection]) - , apnSessionConnectionInfo :: !ApnConnectionInfo - , apnSessionConnectionManager :: !ThreadId - , apnSessionOpen :: !(IORef Bool)} + { apnSessionPool :: !(Pool ApnConnection) + , apnSessionOpen :: !(IORef Bool) + } -- | Information about an APN connection data ApnConnectionInfo = ApnConnectionInfo @@ -110,7 +113,6 @@ { apnConnectionConnection :: !Http2Client , apnConnectionInfo :: !ApnConnectionInfo , apnConnectionWorkerPool :: !QSem - , apnConnectionLastUsed :: !Int64 , apnConnectionFlowControlWorker :: !ThreadId , apnConnectionOpen :: !(IORef Bool)} @@ -151,11 +153,9 @@ | ApnMessageResultFatalError ApnFatalError | ApnMessageResultTemporaryError ApnTemporaryError | ApnMessageResultIOError IOError + | ApnMessageResultClientError ClientError deriving (Eq, Show) -instance SpecifyError ApnMessageResult where - isAnError = ApnMessageResultIOError - -- | The specification of a push notification's message body data JsonApsAlert = JsonApsAlert { jaaTitle :: !(Maybe Text) @@ -377,21 +377,31 @@ -- ^ True if the apn development servers should be used, False to use the production servers -> Int -- ^ How many messages will be sent in parallel? This corresponds to the number of http2 streams open in parallel; 100 seems to be a default value. + -> Int + -- ^ How many connections to be opened at maximum. -> ByteString -- ^ Topic (bundle name of the app) -> IO ApnSession -- ^ The newly created session -newSession certKey certPath caPath dev maxparallel topic = do +newSession certKey certPath caPath dev maxparallel maxConnectionCount topic = do let hostname = if dev then "api.development.push.apple.com" else "api.push.apple.com" connInfo = ApnConnectionInfo certPath certKey caPath hostname maxparallel topic certsOk <- checkCertificates connInfo unless certsOk $ error "Unable to load certificates and/or the private key" - connections <- newIORef [] - connectionManager <- forkIO $ manage 600 connections isOpen <- newIORef True - let session = ApnSession connections connInfo connectionManager isOpen + + let connectionUnusedTimeout :: NominalDiffTime + connectionUnusedTimeout = 600 + pool <- + createPool + (newConnection connInfo) closeApnConnection 1 connectionUnusedTimeout maxConnectionCount + let session = + ApnSession + { apnSessionPool = pool + , apnSessionOpen = isOpen + } addFinalizer session $ closeSession session return session @@ -406,41 +416,25 @@ closeSession s = do isOpen <- atomicModifyIORef' (apnSessionOpen s) (False,) unless isOpen $ error "Session is already closed" - killThread (apnSessionConnectionManager s) - let ioref = apnSessionPool s - openConnections <- atomicModifyIORef' ioref ([],) - mapM_ closeApnConnection openConnections + destroyAllResources (apnSessionPool s) -- | Check whether a session is open or has been closed -- by a call to closeSession isOpen :: ApnSession -> IO Bool isOpen = readIORef . apnSessionOpen -withConnection :: ApnSession -> (ApnConnection -> IO a) -> IO a +withConnection :: ApnSession -> (ApnConnection -> ClientIO a) -> ClientIO a withConnection s action = do - ensureOpen s - let pool = apnSessionPool s - connections <- readIORef pool - let len = length connections - if len == 0 - then do - conn <- newConnection s - res <- action conn - atomicModifyIORef' pool (\a -> (conn:a, ())) - return res - else do - num <- randomRIO (0, len - 1) - currtime <- round <$> getPOSIXTime :: IO Int64 - let conn = connections !! num - conn1 = conn { apnConnectionLastUsed=currtime } - atomicModifyIORef' pool (\a -> (removeNth num a, ())) - isOpen <- readIORef (apnConnectionOpen conn) - if isOpen - then do - res <- action conn1 - atomicModifyIORef' pool (\a -> (conn1:a, ())) - return res - else withConnection s action + lift $ ensureOpen s + ExceptT . try $ + withResource (apnSessionPool s) $ \conn -> do + res <- runClientIO (action conn) + case res of + Left clientError -> + -- When there is a clientError, we think that the connetion is broken. + -- Throwing an exception is the way we inform the resource pool. + throw clientError + Right res -> return res checkCertificates :: ApnConnectionInfo -> IO Bool checkCertificates aci = do @@ -448,26 +442,8 @@ credential <- credentialLoadX509 (aciCertPath aci) (aciCertKey aci) return $ isJust castore && isRight credential -replaceNth n newVal (x:xs) - | n == 0 = newVal:xs - | otherwise = x:replaceNth (n-1) newVal xs - -removeNth n (x:xs) - | n == 0 = xs - | otherwise = x:removeNth (n-1) xs - -manage :: Int64 -> IORef [ApnConnection] -> IO () -manage timeout ioref = forever $ do - currtime <- round <$> getPOSIXTime :: IO Int64 - let minTime = currtime - timeout - expiredOnes <- atomicModifyIORef' ioref - (foldl ( \(a,b) i -> if apnConnectionLastUsed i < minTime then (a, i:b ) else ( i:a ,b)) ([],[])) - mapM_ closeApnConnection expiredOnes - threadDelay 60000000 - -newConnection :: ApnSession -> IO ApnConnection -newConnection apnSession = do - let aci = apnSessionConnectionInfo apnSession +newConnection :: ApnConnectionInfo -> IO ApnConnection +newConnection aci = do Just castore <- readCertificateStore $ aciCaPath aci Right credential <- credentialLoadX509 (aciCertPath aci) (aciCertKey aci) let credentials = Credentials [credential] @@ -496,27 +472,32 @@ ] hostname = aciHostname aci - httpFrameConnection <- newHttp2FrameConnection (T.unpack hostname) 443 (Just clip) isOpen <- newIORef True let handleGoAway rsgaf = do - writeIORef isOpen False + lift $ writeIORef isOpen False return () - client <- newHttp2Client httpFrameConnection 4096 4096 conf handleGoAway ignoreFallbackHandler - linkAsyncs client + client <- + fmap (either throw id) . runClientIO $ do + httpFrameConnection <- newHttp2FrameConnection (T.unpack hostname) 443 (Just clip) + client <- + newHttp2Client httpFrameConnection 4096 4096 conf handleGoAway ignoreFallbackHandler + linkAsyncs client + return client flowWorker <- forkIO $ forever $ do - updated <- _updateWindow $ _incomingFlowControl client + updated <- runClientIO $ _updateWindow $ _incomingFlowControl client threadDelay 1000000 - workersem <- newQSem maxConcurrentStreams - currtime <- round <$> getPOSIXTime :: IO Int64 - return $ ApnConnection client aci workersem currtime flowWorker isOpen + return $ ApnConnection client aci workersem flowWorker isOpen closeApnConnection :: ApnConnection -> IO () -closeApnConnection connection = do - writeIORef (apnConnectionOpen connection) False +closeApnConnection connection = + -- Ignoring ClientErrors in this place. We want to close our session, so we do not need to + -- fail on this kind of errors. + void $ runClientIO $ do + lift $ writeIORef (apnConnectionOpen connection) False let flowWorker = apnConnectionFlowControlWorker connection - killThread flowWorker + lift $ killThread flowWorker _gtfo (apnConnectionConnection connection) HTTP2.NoError "" _close (apnConnectionConnection connection) @@ -531,7 +512,7 @@ -- ^ The message to send -> IO ApnMessageResult -- ^ The response from the APN server -sendRawMessage s token payload = catchIOErrors $ +sendRawMessage s token payload = catchErrors $ withConnection s $ \c -> sendApnRaw c token payload @@ -545,7 +526,7 @@ -- ^ The message to send -> IO ApnMessageResult -- ^ The response from the APN server -sendMessage s token payload = catchIOErrors $ +sendMessage s token payload = catchErrors $ withConnection s $ \c -> sendApnRaw c token message where message = L.toStrict $ encode payload @@ -558,7 +539,7 @@ -- ^ Device to send the message to -> IO ApnMessageResult -- ^ The response from the APN server -sendSilentMessage s token = catchIOErrors $ +sendSilentMessage s token = catchErrors $ withConnection s $ \c -> sendApnRaw c token message where message = "{\"aps\":{\"content-available\":1}}" @@ -576,10 +557,10 @@ -- ^ Device to send the message to -> ByteString -- ^ The message to send - -> IO ApnMessageResult + -> ClientIO ApnMessageResult sendApnRaw connection token message = bracket_ - (waitQSem (apnConnectionWorkerPool connection)) - (signalQSem (apnConnectionWorkerPool connection)) $ do + (lift $ waitQSem (apnConnectionWorkerPool connection)) + (lift $ signalQSem (apnConnectionWorkerPool connection)) $ do let requestHeaders = [ ( ":method", "POST" ) , ( ":scheme", "https" ) , ( ":authority", TE.encodeUtf8 hostname ) @@ -597,7 +578,7 @@ -- sendData client stream (HTTP2.setEndStream) message upload message (HTTP2.setEndHeader . HTTP2.setEndStream) client (_outgoingFlowControl client) stream osfc let pph hStreamId hStream hHeaders hIfc hOfc = - print hHeaders + lift $ print hHeaders response <- waitStream stream isfc pph let (errOrHeaders, frameResponses, _) = response case errOrHeaders of @@ -632,8 +613,17 @@ getHeaderEx :: HTTP2.HeaderName -> [HTTP2.Header] -> HTTP2.HeaderValue getHeaderEx name headers = fromMaybe (throw $ ApnExceptionMissingHeader name) (DL.lookup name headers) -catchIOErrors :: SpecifyError a => IO a -> IO a -catchIOErrors = flip catchIOError (return . isAnError) + +catchErrors :: ClientIO ApnMessageResult -> IO ApnMessageResult +catchErrors = catchIOErrors . catchClientErrors + where + catchIOErrors :: IO ApnMessageResult -> IO ApnMessageResult + catchIOErrors = flip catchIOError (return . ApnMessageResultIOError) + + catchClientErrors :: ClientIO ApnMessageResult -> IO ApnMessageResult + catchClientErrors act = + either ApnMessageResultClientError id <$> runClientIO act + -- The type of permanent error indicated by APNS -- See https://apple.co/2RDCdWC table 8-6 for the meaning of each value.