diff --git a/pubnub.cabal b/pubnub.cabal --- a/pubnub.cabal +++ b/pubnub.cabal @@ -2,7 +2,7 @@ -- see http://haskell.org/cabal/users-guide/ name: pubnub -version: 1.0.0 +version: 1.1.0 synopsis: PubNub Haskell SDK description: The PubNub Network makes Real-time easy with a Simple Communications API. Two Functions: Send/Receive (Publish/Subscribe). We provide a web-service API for businesses to build scalable Data Push communication apps on Mobile, Tablet and Web. Bidirectional JSON. homepage: http://github.com/pubnub/haskell @@ -22,9 +22,10 @@ -- other-extensions: build-depends: base >=4.6 && <4.7 , aeson >= 0.7.0.0 + , conduit >= 1.1.0.2 + , conduit-extra >= 1.1.0 , http-conduit >= 2.0.0.3 - , http-client >= 0.2.0.3 - , conduit >= 1.0.9.3 + , http-client >= 0.2.0.3 , bytestring >= 0.10.0.2 , transformers >= 0.3.0.0 , vector >= 0.10.9.1 @@ -39,7 +40,7 @@ , cipher-aes >= 0.2.6 , crypto-api >= 0.12.2.2 , crypto-cipher-types >= 0.0.9 - , base64-bytestring >= 1.0.0.1 + , base64-bytestring >= 1.0.0.1 , time >= 1.4.0.1 ghc-options: -Wall hs-source-dirs: src @@ -95,4 +96,4 @@ , tasty , tasty-hunit , tasty-quickcheck - , tasty-smallcheck + , tasty-smallcheck diff --git a/src/Network/Pubnub.hs b/src/Network/Pubnub.hs --- a/src/Network/Pubnub.hs +++ b/src/Network/Pubnub.hs @@ -31,6 +31,8 @@ import Data.UUID.V4 import Network.HTTP.Conduit import Network.HTTP.Types +import Data.Conduit + import Control.Monad.Trans import Control.Concurrent.Async import Control.Applicative ((<$>)) @@ -73,13 +75,13 @@ subscribeInternal :: (FromJSON b) => PN -> SubscribeOptions b -> IO (Async ()) subscribeInternal pn subOpts = - async (withManager $ \manager -> lift $ connect pn{time_token = Timestamp 0} manager False) + async (withManagerSettings conduitManagerSettings $ \manager -> connect pn{time_token = Timestamp 0} manager False) where connect pn' manager isReconnect = do let req = buildSubscribeRequest pn' "0" - eres <- try $ httpLbs req manager :: IO (Either HttpException (Response L.ByteString)) + eres <- try $ httpLbs req manager case eres of - Right res -> + Right res -> do case decode $ responseBody res of Just (ConnectResponse ([], t)) -> do liftIO (if isReconnect @@ -99,25 +101,16 @@ connect pn' manager isReconnect subscribe' manager pn' = do - let req = buildSubscribeRequest pn' $ head . L.toChunks $ encode (time_token pn') - eres <- try $ httpLbs req manager :: IO (Either HttpException (Response L.ByteString)) + let req = buildSubscribeRequest pn' $ L.toStrict $ encode (time_token pn') + eres <- try $ http req manager case eres of - Right res -> + Right res -> do case (ctx pn', iv pn') of - (Just c, Just i) -> - case decode $ responseBody res of - Just (EncryptedSubscribeResponse (resp, t)) -> do - _ <- liftIO $ mapM (onMsg subOpts . decodeEncrypted c i) resp - subscribe' manager (pn' { time_token=t }) - Nothing -> - subscribe' manager pn' - (_, _) -> - case decode $ responseBody res of - Just (SubscribeResponse (resp, t)) -> do - _ <- liftIO $ mapM (onMsg subOpts) resp - subscribe' manager (pn' { time_token=t }) - Nothing -> - subscribe' manager pn' + (Just c, Just i) -> do + responseBody res $$+- encryptedSubscribeSink c i + (_, _) -> do + responseBody res $$+- subscribeSink + reconnect pn' manager Left (ResponseTimeout :: HttpException) -> subscribe' manager pn' Left (StatusCodeException (Status code msg) _ _) -> do @@ -127,10 +120,28 @@ liftIO $ onError subOpts Nothing Nothing reconnect pn' manager + encryptedSubscribeSink c i = + awaitForever $ (\x -> + case decode (L.fromStrict x) of + Just (EncryptedSubscribeResponse (resp, _)) -> do + _ <- liftIO $ mapM (onMsg subOpts . decodeEncrypted c i) resp + return () + Nothing -> + return ()) + + subscribeSink = + awaitForever $ (\x -> + case decode (L.fromStrict x) of + Just (SubscribeResponse (resp, _)) -> do + _ <- liftIO $ mapM (onMsg subOpts) resp + return () + Nothing -> + return ()) + reconnect pn' manager = connect pn' manager True buildSubscribeRequest pn' tt = - buildRequest pn' [ "subscribe" + buildRequest pn' [ "stream" , encodeUtf8 $ sub_key pn' , encodeUtf8 $ T.intercalate "," (channels pn') , bsFromInteger $ jsonp_callback pn' @@ -152,7 +163,7 @@ publish :: ToJSON a => PN -> T.Text -> a -> IO (Maybe PublishResponse) publish pn channel msg = do - let encoded_msg = head . L.toChunks $ encode msg + let encoded_msg = L.toStrict $ encode msg let sig = signature (sec_key pn) encoded_msg let req = buildRequest pn [ "publish" , encodeUtf8 $ pub_key pn @@ -242,21 +253,21 @@ auth pn k = pn{auth_key = (Just k)} pamDo :: B.ByteString -> PN -> Auth -> IO (Maybe Value) -pamDo pamMethod pn authR = do - ts <- (bsFromInteger . round) <$> getPOSIXTime - let req = buildRequest pn pamURI ((pamQS ts) ++ [("signature", signature ts)]) Nothing +pamDo pamMethod pn authR = do + ts <- (bsFromInteger . round) <$> getPOSIXTime + let req = buildRequest pn pamURI ((pamQS ts) ++ [("signature", signature ts)]) Nothing res <- withManager $ httpLbs req - return (decode $ responseBody res) + return (decode $ responseBody res) where authK = T.intercalate "," $ authKeys authR channel = fromJust $ chan authR authRead = r authR authWrite = w authR - + pubKey = encodeUtf8 $ pub_key pn subKey = encodeUtf8 $ sub_key pn secKey = L.fromStrict . encodeUtf8 $ sec_key pn - + signature ts = B64.encode . L.toStrict . bytestringDigest $ hmacSha256 secKey (msg ts) msg ts = L.fromStrict $ B.intercalate B.empty [ subKey, "\n" @@ -265,7 +276,7 @@ , B.tail $ renderSimpleQuery True $ pamQS ts] pamURI = [ "v1" - , "auth" + , "auth" , pamMethod , "sub-key" , subKey] @@ -292,7 +303,7 @@ , requestHeaders = [ ("V", "3.1") , ("User-Agent", "Haskell") , ("Accept", "*/*")] - , queryString = renderSimpleQuery True qs + , queryString = renderSimpleQuery True qs , secure = ssl pn , responseTimeout = Just (maybe 5000000 (* 1000000) timeout) }