diff --git a/LICENSE b/LICENSE --- a/LICENSE +++ b/LICENSE @@ -1,4 +1,4 @@ -Copyright Athen Clark (c) 2019 +Copyright Athen Clark (c) 2019, 2020 All rights reserved. diff --git a/src/Test/Serialization/Symbiote.hs b/src/Test/Serialization/Symbiote.hs --- a/src/Test/Serialization/Symbiote.hs +++ b/src/Test/Serialization/Symbiote.hs @@ -83,9 +83,14 @@ -} module Test.Serialization.Symbiote - ( SymbioteOperation (..), Symbiote (..), SimpleSerialization (..), Topic, SymbioteT, register - , firstPeer, secondPeer, First (..), Second (..), Generating (..), Operating (..), Failure (..) - , defaultSuccess, defaultFailure, defaultProgress, nullProgress, simpleTest, simpleTest' + ( -- * Suite Building + SymbioteOperation (..), Symbiote (..), SimpleSerialization (..), Topic, SymbioteT, register + , -- * Protocol Messages + First (..), Second (..), Generating (..), Operating (..) + , -- * Result Handling + Failure (..), defaultSuccess, defaultFailure, defaultProgress, nullProgress + , -- * Test Execution + simpleTest, simpleTest', firstPeer, secondPeer ) where import Test.Serialization.Symbiote.Core @@ -94,6 +99,7 @@ import Data.Map (Map) import qualified Data.Map as Map +import Data.Set (Set) import qualified Data.Set as Set import qualified Data.ByteString as BS import qualified Data.ByteString.Lazy as LBS @@ -103,8 +109,8 @@ import Data.Aeson (ToJSON (..), FromJSON (..), (.=), object, (.:), Value (Object, String)) import Data.Aeson.Types (typeMismatch) import Data.Serialize (Serialize (..)) -import Data.Serialize.Put (putWord8, putInt32be, putByteString, putLazyByteString) -import Data.Serialize.Get (getWord8, getInt32be, getByteString, getLazyByteString) +import Data.Serialize.Put (putWord8, putInt32be, putByteString, putLazyByteString, PutM) +import Data.Serialize.Get (getWord8, getInt32be, getByteString, getLazyByteString, Get) import Text.Printf (printf) import Control.Concurrent.STM (TVar, newTVarIO, readTVarIO, writeTVar, atomically, newTChan, readTChan, writeTChan) @@ -120,10 +126,14 @@ -- | The most trivial serialization medium for any @a@ and @o@. +-- There's no need to implement transmission protocol specific instances for this type, like 'ToJSON' or 'Serialize', because it is intended to operate locally (on the same program), and over 'Eq'. data SimpleSerialization a o - = SimpleValue a - | SimpleOutput o - | SimpleOperation (Operation a) + = -- | A value @a@ encodes as just that value + SimpleValue a + | -- | An output @o@ encodes as just that output + SimpleOutput o + | -- | An operation 'Operation' encodes as just that operation + SimpleOperation (Operation a) deriving (Generic) deriving instance (Show a, Show o, Show (Operation a)) => Show (SimpleSerialization a o) deriving instance (Eq a, Eq o, Eq (Operation a)) => Eq (SimpleSerialization a o) @@ -140,16 +150,16 @@ decodeOp _ = Nothing --- | Register a topic in the test suite +-- | Register a topic in the test suite builder. register :: forall a o s m . Arbitrary a => Arbitrary (Operation a) => Symbiote a o s => Eq o => MonadIO m - => Topic + => Topic -- ^ Topic name as a 'Text' -> Int32 -- ^ Max size - -> Proxy a -- ^ Reference to datatype + -> Proxy a -- ^ Reference to the datatype -> SymbioteT s m () register t maxSize Proxy = do generation <- liftIO (newTVarIO newGeneration) @@ -170,16 +180,25 @@ } modify' (Map.insert t (ExistsSymbiote newState)) --- | Messages sent by a peer during their generating phase + +---------------- Protocol Messages ----------------------- + + +-- | Messages sent by a peer during their generating phase - polymorphic in the serialization medium. data Generating s - = Generated + = -- | \"I\'ve generated a value and operation, here you go.\" + Generated { genValue :: s , genOperation :: s } - | BadResult s -- ^ Expected value - | YourTurn - | ImFinished - | GeneratingNoParseOperated s + | -- | \"You sent the wrong value!\" + BadResult s + | -- | \"It\'s your turn to generate, I just finished and we\'re both O.K.\" + YourTurn + | -- | \"I just finished all generation, and my topic state\'s 'size' is equal to the 'maxSize'.\" + ImFinished + | -- | \"I could not deserialize the output value sent by you, and I have to tell you about it.\" + GeneratingNoParseOperated s deriving (Eq, Show, Generic) instance Arbitrary s => Arbitrary (Generating s) where arbitrary = oneof @@ -244,11 +263,14 @@ 4 -> GeneratingNoParseOperated <$> getLazyByteString' _ -> fail "Generating LazyByteString" --- | Messages sent by a peer during their operating phase +-- | Messages sent by a peer during their operating phase - polymorphic in the serialization medium. data Operating s - = Operated s -- ^ Serialized value after operation - | OperatingNoParseValue s - | OperatingNoParseOperation s + = -- | \"I\'ve performed the operation on the value, and here's the output result.\" + Operated s + | -- | \"I couldn't deserialize the input value sent by you, and I have to tell you about it.\" + OperatingNoParseValue s + | -- | \"I couldn't deserialize the operation sent by you, and I have to tell you about it.\" + OperatingNoParseOperation s deriving (Eq, Show, Generic) instance Arbitrary s => Arbitrary (Operating s) where arbitrary = oneof @@ -295,14 +317,19 @@ 2 -> OperatingNoParseOperation <$> getLazyByteString' _ -> fail "Operating LazyByteString" --- | Messages sent by the first peer +-- | Messages sent by the first peer - polymorphic in the serialization medium. data First s - = AvailableTopics (Map Topic Int32) -- ^ Mapping of topics to their gen size - | FirstGenerating + = -- | \"Here are the topics I support.\" + AvailableTopics (Map Topic Int32) + | -- | \"I got your subset of topics, but they don\'t match to mine.\" + BadStartSubset + | -- | \"It\'s my turn to generate, so here\'s a generating message.\" + FirstGenerating { firstGeneratingTopic :: Topic , firstGenerating :: Generating s } - | FirstOperating + | -- | \"It's my turn to operate, so here\'s my operating message.\" + FirstOperating { firstOperatingTopic :: Topic , firstOperating :: Operating s } @@ -310,25 +337,33 @@ instance Arbitrary s => Arbitrary (First s) where arbitrary = oneof [ AvailableTopics <$> arbitrary + , pure BadStartSubset , FirstGenerating <$> arbitrary <*> arbitrary , FirstOperating <$> arbitrary <*> arbitrary ] instance ToJSON s => ToJSON (First s) where toJSON x = case x of AvailableTopics ts -> object ["availableTopics" .= ts] + BadStartSubset -> String "badStartSubset" FirstGenerating t y -> object ["firstGenerating" .= object ["topic" .= t, "generating" .= y]] FirstOperating t y -> object ["firstOperating" .= object ["topic" .= t, "operating" .= y]] instance FromJSON s => FromJSON (First s) where - parseJSON (Object o) = availableTopics <|> firstGenerating' <|> firstOperating' + parseJSON json = case json of + Object o -> + let availableTopics = AvailableTopics <$> o .: "availableTopics" + firstGenerating' = do + o' <- o .: "firstGenerating" + FirstGenerating <$> o' .: "topic" <*> o' .: "generating" + firstOperating' = do + o' <- o .: "firstOperating" + FirstOperating <$> o' .: "topic" <*> o' .: "operating" + in availableTopics <|> firstGenerating' <|> firstOperating' + String s + | s == "badStartSubset" -> pure BadStartSubset + | otherwise -> fail' + _ -> fail' where - availableTopics = AvailableTopics <$> o .: "availableTopics" - firstGenerating' = do - o' <- o .: "firstGenerating" - FirstGenerating <$> o' .: "topic" <*> o' .: "generating" - firstOperating' = do - o' <- o .: "firstOperating" - FirstOperating <$> o' .: "topic" <*> o' .: "operating" - parseJSON x = typeMismatch "First s" x + fail' = typeMismatch "First s" json instance Serialize (First BS.ByteString) where put x = case x of AvailableTopics ts -> do @@ -336,17 +371,19 @@ let ls = Map.toList ts putInt32be (fromIntegral (length ls)) void (traverse put ls) - FirstGenerating t y -> putWord8 1 *> put t *> put y - FirstOperating t y -> putWord8 2 *> put t *> put y + BadStartSubset -> putWord8 1 + FirstGenerating t y -> putWord8 2 *> put t *> put y + FirstOperating t y -> putWord8 3 *> put t *> put y get = do x <- getWord8 case x of 0 -> do l <- getInt32be AvailableTopics . Map.fromList <$> replicateM (fromIntegral l) get - 1 -> FirstGenerating <$> get <*> get - 2 -> FirstOperating <$> get <*> get - _ -> fail "First s" + 1 -> pure BadStartSubset + 2 -> FirstGenerating <$> get <*> get + 3 -> FirstOperating <$> get <*> get + _ -> fail "First BS.ByteString" instance Serialize (First LBS.ByteString) where put x = case x of AvailableTopics ts -> do @@ -354,38 +391,47 @@ let ls = Map.toList ts putInt32be (fromIntegral (length ls)) void (traverse put ls) - FirstGenerating t y -> putWord8 1 *> put t *> put y - FirstOperating t y -> putWord8 2 *> put t *> put y + BadStartSubset -> putWord8 1 + FirstGenerating t y -> putWord8 2 *> put t *> put y + FirstOperating t y -> putWord8 3 *> put t *> put y get = do x <- getWord8 case x of 0 -> do l <- getInt32be AvailableTopics . Map.fromList <$> replicateM (fromIntegral l) get - 1 -> FirstGenerating <$> get <*> get - 2 -> FirstOperating <$> get <*> get - _ -> fail "First s" + 1 -> pure BadStartSubset + 2 -> FirstGenerating <$> get <*> get + 3 -> FirstOperating <$> get <*> get + _ -> fail "First LBS.ByteString" +-- | Convenience function for avoiding a case match over First getFirstGenerating :: First s -> Maybe (Topic, Generating s) getFirstGenerating x = case x of FirstGenerating topic g -> Just (topic, g) _ -> Nothing +-- | Convenience function for avoiding a case match over First getFirstOperating :: First s -> Maybe (Topic, Operating s) getFirstOperating x = case x of FirstOperating topic g -> Just (topic, g) _ -> Nothing --- | Messages sent by the second peer +-- | Messages sent by the second peer - polymorphic in the serialization medium. data Second s - = BadTopics (Map Topic Int32) - | Start - | SecondOperating + = -- | \"Although my topics should be at least a subset of your topics available, + -- the following of mine do not have the same max size as yours.\" + BadTopics (Map Topic Int32) + | -- | \"All systems nominal, you may fire (the following subset of topics) when ready.\" + Start (Set Topic) + | -- | \"It's my turn to operate, so here\'s my operating message.\" + SecondOperating { secondOperatingTopic :: Topic , secondOperating :: Operating s } - | SecondGenerating + | -- | \"It's my turn to generate, so here\'s my generating message.\" + SecondGenerating { secondGeneratingTopic :: Topic , secondGenerating :: Generating s } @@ -393,29 +439,27 @@ instance Arbitrary s => Arbitrary (Second s) where arbitrary = oneof [ BadTopics <$> arbitrary - , pure Start + , Start <$> arbitrary , SecondOperating <$> arbitrary <*> arbitrary , SecondGenerating <$> arbitrary <*> arbitrary ] instance ToJSON s => ToJSON (Second s) where toJSON x = case x of BadTopics ts -> object ["badTopics" .= ts] - Start -> String "start" + Start ts -> object ["start" .= ts] SecondOperating t y -> object ["secondOperating" .= object ["topic" .= t, "operating" .= y]] SecondGenerating t y -> object ["secondGenerating" .= object ["topic" .= t, "generating" .= y]] instance FromJSON s => FromJSON (Second s) where - parseJSON (Object o) = badTopics <|> secondOperating' <|> secondGenerating' + parseJSON (Object o) = badTopics <|> start <|> secondOperating' <|> secondGenerating' where badTopics = BadTopics <$> o .: "badTopics" + start = Start <$> o .: "start" secondOperating' = do o' <- o .: "secondOperating" SecondOperating <$> o' .: "topic" <*> o' .: "operating" secondGenerating' = do o' <- o .: "secondGenerating" SecondGenerating <$> o' .: "topic" <*> o' .: "generating" - parseJSON x@(String s) - | s == "start" = pure Start - | otherwise = typeMismatch "Second s" x parseJSON x = typeMismatch "Second s" x instance Serialize (Second BS.ByteString) where put x = case x of @@ -424,7 +468,11 @@ let ls = Map.toList ts putInt32be (fromIntegral (length ls)) void (traverse put ls) - Start -> putWord8 1 + Start ts -> do + putWord8 1 + let ls = Set.toList ts + putInt32be (fromIntegral (length ls)) + void (traverse put ls) SecondOperating t y -> putWord8 2 *> put t *> put y SecondGenerating t y -> putWord8 3 *> put t *> put y get = do @@ -433,10 +481,12 @@ 0 -> do l <- getInt32be BadTopics . Map.fromList <$> replicateM (fromIntegral l) get - 1 -> pure Start + 1 -> do + l <- getInt32be + Start . Set.fromList <$> replicateM (fromIntegral l) get 2 -> SecondOperating <$> get <*> get 3 -> SecondGenerating <$> get <*> get - _ -> fail "Second s" + _ -> fail "Second BS.ByteString" instance Serialize (Second LBS.ByteString) where put x = case x of BadTopics ts -> do @@ -444,7 +494,11 @@ let ls = Map.toList ts putInt32be (fromIntegral (length ls)) void (traverse put ls) - Start -> putWord8 1 + Start ts -> do + putWord8 1 + let ls = Set.toList ts + putInt32be (fromIntegral (length ls)) + void (traverse put ls) SecondOperating t y -> putWord8 2 *> put t *> put y SecondGenerating t y -> putWord8 3 *> put t *> put y get = do @@ -453,31 +507,43 @@ 0 -> do l <- getInt32be BadTopics . Map.fromList <$> replicateM (fromIntegral l) get - 1 -> pure Start + 1 -> do + l <- getInt32be + Start . Set.fromList <$> replicateM (fromIntegral l) get 2 -> SecondOperating <$> get <*> get 3 -> SecondGenerating <$> get <*> get - _ -> fail "Second s" + _ -> fail "Second LBS.ByteString" +-- | Convenience function for avoiding a case match over First getSecondGenerating :: Second s -> Maybe (Topic, Generating s) getSecondGenerating x = case x of SecondGenerating topic g -> Just (topic, g) _ -> Nothing +-- | Convenience function for avoiding a case match over First getSecondOperating :: Second s -> Maybe (Topic, Operating s) getSecondOperating x = case x of SecondOperating topic g -> Just (topic, g) _ -> Nothing +-- | Exception data type data Failure them s - = BadTopicsFailure + = -- | Topic sets do not match between peers + BadTopicsFailure { badTopicsFirst :: Map Topic Int32 , badTopicsSecond :: Map Topic Int32 } - | OutOfSyncFirst (First s) - | OutOfSyncSecond (Second s) - | TopicNonexistent Topic - | WrongTopic + | -- | The first peer doesn\'t have the subset topics identified by second + BadStartSubsetFailure (Set Topic) + | -- | The first peer is out of sync and not sending the correct message + OutOfSyncFirst (First s) + | -- | The second peer is out of sync and not sending the correct message + OutOfSyncSecond (Second s) + | -- | Topic does not exist + TopicNonexistent Topic + | -- | Got the wrong topic + WrongTopic { wrongTopicExpected :: Topic , wrongTopicGot :: Topic } @@ -486,10 +552,14 @@ | CantParseGeneratedOperation Topic s | CantParseLocalValue Topic s | CantParseLocalOperation Topic s - | BadOperating Topic (Operating s) - | BadGenerating Topic (Generating s) - | BadThem Topic (them s) - | SafeFailure + | -- | Incorrect operating message received + BadOperating Topic (Operating s) + | -- | Incorrect generating message received + BadGenerating Topic (Generating s) + | -- | Incorrect peer message ('First' and 'Second' agnostic) + BadThem Topic (them s) + | -- | Failed because the output of the operation applied to the value does not match between the peers. + SafeFailure { safeFailureTopic :: Topic , safeFailureExpected :: s , safeFailureGot :: s @@ -497,15 +567,15 @@ deriving (Eq, Show) --- | Via putStrLn +-- | Via 'putStrLn' defaultSuccess :: Topic -> IO () defaultSuccess (Topic t) = putStrLn $ "Topic " ++ unpack t ++ " succeeded" --- | Via putStrLn +-- | Via 'putStrLn' defaultFailure :: Show (them s) => Show s => Failure them s -> IO () defaultFailure f = error $ "Failure: " ++ show f --- | Via putStrLn +-- | Via 'putStrLn' defaultProgress :: Topic -> Float -> IO () defaultProgress (Topic t) p = putStrLn $ "Topic " ++ unpack t ++ ": " ++ printf "%.2f" (p * 100.0) ++ "%" @@ -514,7 +584,7 @@ nullProgress _ _ = pure () --- | Run the test suite as the first peer - see 'Test.Serialization.Symbiote.WebSocket' for end-user +-- | Run the test suite as the first peer - see "Test.Serialization.Symbiote.WebSocket" and "Test.Serialization.Symbiote.ZeroMQ" for end-user -- implementations. firstPeer :: forall m s . MonadIO m @@ -528,42 +598,46 @@ -> m () firstPeer encodeAndSend receiveAndDecode onSuccess onFailure onProgress x = do state <- runSymbioteT x True - let topics = go <$> state + let myTopics = go <$> state where go s = case s of ExistsSymbiote s' -> maxSize s' - encodeAndSend (AvailableTopics topics) + encodeAndSend (AvailableTopics myTopics) shouldBeStart <- receiveAndDecode case shouldBeStart of - BadTopics badTopics -> onFailure $ BadTopicsFailure topics badTopics - Start -> do - topicsToProcess <- liftIO (newTVarIO (Map.keysSet topics)) - let processAllTopics = do - mTopicToProcess <- Set.maxView <$> liftIO (readTVarIO topicsToProcess) - case mTopicToProcess of - Nothing -> pure () -- done - Just (topic, newTopics) -> do - liftIO (atomically (writeTVar topicsToProcess newTopics)) - case Map.lookup topic state of - Nothing -> onFailure $ TopicNonexistent topic - Just symbioteState -> do - hasSentFinishedVar <- liftIO $ newTVarIO HasntSentFinished - hasReceivedFinishedVar <- liftIO $ newTVarIO HasntReceivedFinished - generating - encodeAndSend receiveAndDecode - FirstGenerating FirstOperating - getSecondGenerating getSecondOperating - hasSentFinishedVar hasReceivedFinishedVar - processAllTopics - onSuccess - onFailure - onProgress - topic symbioteState - processAllTopics - _ -> onFailure $ OutOfSyncSecond shouldBeStart + BadTopics badTopics -> onFailure (BadTopicsFailure myTopics badTopics) + Start topicsSubset + | not (topicsSubset `Set.isSubsetOf` Map.keysSet myTopics) -> do + encodeAndSend BadStartSubset + onFailure (BadStartSubsetFailure topicsSubset) + | otherwise -> do + topicsToProcess <- liftIO (newTVarIO topicsSubset) + let processAllTopics = do + mTopicToProcess <- Set.maxView <$> liftIO (readTVarIO topicsToProcess) + case mTopicToProcess of + Nothing -> pure () -- done + Just (topic, newTopics) -> do + liftIO (atomically (writeTVar topicsToProcess newTopics)) + case Map.lookup topic state of + Nothing -> onFailure (TopicNonexistent topic) + Just symbioteState -> do + hasSentFinishedVar <- liftIO (newTVarIO HasntSentFinished) + hasReceivedFinishedVar <- liftIO (newTVarIO HasntReceivedFinished) + generating + encodeAndSend receiveAndDecode + FirstGenerating FirstOperating + getSecondGenerating getSecondOperating + hasSentFinishedVar hasReceivedFinishedVar + processAllTopics + onSuccess + onFailure + onProgress + topic symbioteState + processAllTopics + _ -> onFailure (OutOfSyncSecond shouldBeStart) --- | Run the test suite as the second peer - see 'Test.Serialization.Symbiote.WebSocket' for end-user +-- | Run the test suite as the second peer - see "Test.Serialization.Symbiote.WebSocket" and "Test.Serialization.Symbiote.ZeroMQ" for end-user -- implementations. secondPeer :: forall s m . MonadIO m @@ -579,18 +653,18 @@ state <- runSymbioteT x False shouldBeAvailableTopics <- receiveAndDecode case shouldBeAvailableTopics of - AvailableTopics topics -> do + AvailableTopics topicsAvailable -> do let myTopics = go <$> state where go s = case s of ExistsSymbiote s' -> maxSize s' - if myTopics /= topics + if not (myTopics `Map.isSubmapOf` topicsAvailable) then do encodeAndSend (BadTopics myTopics) - onFailure $ BadTopicsFailure topics myTopics + onFailure (BadTopicsFailure topicsAvailable myTopics) else do - encodeAndSend Start - topicsToProcess <- liftIO (newTVarIO (Map.keysSet topics)) + encodeAndSend (Start (Map.keysSet myTopics)) + topicsToProcess <- liftIO (newTVarIO (Map.keysSet myTopics)) let processAllTopics = do mTopicToProcess <- Set.maxView <$> liftIO (readTVarIO topicsToProcess) case mTopicToProcess of @@ -598,10 +672,10 @@ Just (topic, newTopics) -> do liftIO (atomically (writeTVar topicsToProcess newTopics)) case Map.lookup topic state of - Nothing -> onFailure $ TopicNonexistent topic + Nothing -> onFailure (TopicNonexistent topic) Just symbioteState -> do - hasSentFinishedVar <- liftIO $ newTVarIO HasntSentFinished - hasReceivedFinishedVar <- liftIO $ newTVarIO HasntReceivedFinished + hasSentFinishedVar <- liftIO (newTVarIO HasntSentFinished) + hasReceivedFinishedVar <- liftIO (newTVarIO HasntReceivedFinished) operating encodeAndSend receiveAndDecode SecondGenerating SecondOperating @@ -613,7 +687,7 @@ onProgress topic symbioteState processAllTopics - _ -> onFailure $ OutOfSyncFirst shouldBeAvailableTopics + _ -> onFailure (OutOfSyncFirst shouldBeAvailableTopics) data HasSentFinished @@ -888,19 +962,22 @@ encodeAndSendChan chan x = liftIO $ atomically (writeTChan chan x) receiveAndDecodeChan chan = liftIO $ atomically (readTChan chan) - +putByteString' :: BS.ByteString -> PutM () putByteString' b = do putInt32be (fromIntegral (BS.length b)) putByteString b +getByteString' :: Get BS.ByteString getByteString' = do l <- getInt32be getByteString (fromIntegral l) +putLazyByteString' :: LBS.ByteString -> PutM () putLazyByteString' b = do putInt32be (fromIntegral (LBS.length b)) putLazyByteString b +getLazyByteString' :: Get LBS.ByteString getLazyByteString' = do l <- getInt32be getLazyByteString (fromIntegral l) diff --git a/src/Test/Serialization/Symbiote/Core.hs b/src/Test/Serialization/Symbiote/Core.hs --- a/src/Test/Serialization/Symbiote/Core.hs +++ b/src/Test/Serialization/Symbiote/Core.hs @@ -18,6 +18,12 @@ Maintainer: athan.clark@gmail.com Portability: GHC +This module defines the machinery of any topic's state machine: + +* agnostic for whatever type or operation you're testing +* doesn't try to define the stuff that orchestrates multiple topics operating together +* doesn't specify how the protocol actually looks with respect to the messages sent between the peers + -} module Test.Serialization.Symbiote.Core where @@ -47,14 +53,18 @@ -- | A type-level relation between a type and appropriate, testable operations on that type. class SymbioteOperation a o | a -> o where + -- | An enumerated type of operations on @a@ that result in @o@. data Operation a :: * + -- | Apply the 'Operation' to @a@, to get an @o@. perform :: Operation a -> a -> o -- | A serialization format for a particular type, and serialized data type. class SymbioteOperation a o => Symbiote a o s | a -> o where encode :: a -> s decode :: s -> Maybe a + -- | Needs a reference to @a@ because the fundep is only one direction (i.e. only one output defined per input, but could be used elsewhere) encodeOut :: Proxy a -> o -> s + -- | Needs a reference to @a@ because the fundep is only one direction decodeOut :: Proxy a -> s -> Maybe o encodeOp :: Operation a -> s decodeOp :: s -> Maybe (Operation a) @@ -63,6 +73,7 @@ -- | Unique name of a type, for a suite of tests newtype Topic = Topic Text deriving (Eq, Ord, Show, IsString, Arbitrary, ToJSON, FromJSON, ToJSONKey, FromJSONKey) +-- | Serialized as a @String32@ in the . instance Serialize Topic where put (Topic t) = do putInt32be (fromIntegral (Data.Text.length t)) @@ -71,25 +82,31 @@ l <- getInt32be Topic . pack <$> replicateM (fromIntegral l) get --- | Protocol state for a particular topic +-- | The state-machine representation for a particular topic, identical for each peer. As the protocol progresses and messages are passed, and values are generated, this +-- value will change over time, and is unique for each topic. data SymbioteProtocol a s - = MeGenerated - { meGenValue :: a - , meGenOperation :: Operation a - , meGenReceived :: Maybe s -- ^ Remotely operated value + = -- | \"I\'m generating a value\" + MeGenerated + { meGenValue :: a -- ^ The value generated + , meGenOperation :: Operation a -- ^ The operation generated + , meGenReceived :: Maybe s -- ^ Remotely operated value - might not exist due to the need for the remote party to send the @o@ output value. } - | ThemGenerating - { themGen :: Maybe (s, s) -- ^ Remotely generated value and operation + | -- | \"They\'re generating a value\" + ThemGenerating + { themGen :: Maybe (s, s) -- ^ Remotely generated value and operation - might not exist due to the need for the remote party to send the @a@ and 'Operation' values. } - | NotStarted + | -- | The topic's generation / operation exchange hasn't started for either peer + NotStarted + -- | The topic's generation / operation exchange has completed, and is currently being processed on another topic, or the whole suite is finished for all topics. | Finished --- | Protocol generation state +-- | Protocol state, with amended \"current generation value\" information data SymbioteGeneration a s = SymbioteGeneration - { size :: Int32 - , protocol :: SymbioteProtocol a s + { size :: Int32 -- ^ The current \"size\" for the generated values and operations, with respect to QuickCheck\'s parameter. + , protocol :: SymbioteProtocol a s -- ^ The rest of the state } +-- | The initial state for any topic\'s state machine. newGeneration :: SymbioteGeneration a s newGeneration = SymbioteGeneration { size = 1 @@ -97,24 +114,25 @@ } --- | Internal existential state of a registered topic with type's facilities +-- | Internal existential state of a registered topic, with fields explicitly storing a type's normally typeclass-specified facilities. data SymbioteState a o s = SymbioteState - { generate :: Gen a - , generateOp :: Gen (Operation a) - , equal :: o -> o -> Bool - , maxSize :: Int32 - , generation :: TVar (SymbioteGeneration a s) - , encode' :: a -> s - , encodeOut' :: o -> s - , encodeOp' :: Operation a -> s - , decode' :: s -> Maybe a - , decodeOut' :: s -> Maybe o - , decodeOp' :: s -> Maybe (Operation a) - , perform' :: Operation a -> a -> o + { generate :: Gen a -- ^ Generate a random value + , generateOp :: Gen (Operation a) -- ^ Generate a random operation + , equal :: o -> o -> Bool -- ^ Are the outputs equal? + , maxSize :: Int32 -- ^ Maximum size the internal generation will approach to - with respect to QuickCheck\'s parameter + , generation :: TVar (SymbioteGeneration a s) -- ^ The actual state of the topic, stored as an atomically modifiable 'TVar' + , encode' :: a -> s -- ^ 'encode' + , encodeOut' :: o -> s -- ^ 'encodeOut' + , encodeOp' :: Operation a -> s -- ^ 'encodeOp' + , decode' :: s -> Maybe a -- ^ 'decode' + , decodeOut' :: s -> Maybe o -- ^ 'decodeOut' + , decodeOp' :: s -> Maybe (Operation a) -- ^ 'decodeOp' + , perform' :: Operation a -> a -> o -- ^ 'perform' } +-- | A custom existential quantifier, which places additional typeclass constraints on the @a@ and @o@ variables of 'SymbioteState', leaving only @s@ as the globally visible type variable - the \"serialization\" medium. data ExistsSymbiote s = forall a o . ( Arbitrary a @@ -124,9 +142,10 @@ ) => ExistsSymbiote (SymbioteState a o s) - +-- | Builder for the total set of topics supported by this peer. type SymbioteT s m = ReaderT Bool (StateT (Map Topic (ExistsSymbiote s)) m) +-- | Get the set of topics from the builder. runSymbioteT :: Monad m => SymbioteT s m () -> Bool -- ^ Is this the first peer to initiate the protocol? @@ -134,14 +153,18 @@ runSymbioteT x isFirst = execStateT (runReaderT x isFirst) Map.empty +-- | Used internally - attempt generating a value and operation for a specific topic, by using the mechanics defined in the protocol's stored state and existential typeclass facilities. data GenerateSymbiote s - = DoneGenerating - | GeneratedSymbiote + = -- | Can\'t generate any more values, and \"is /finished/\" + DoneGenerating + | -- | Generated a value and operation, as serialized values + GeneratedSymbiote { generatedValue :: s , generatedOperation :: s } +-- | Given an existentially stored topic state, attempt to generate a value and operation for that topic's type. generateSymbiote :: forall s m. MonadIO m => ExistsSymbiote s -> m (GenerateSymbiote s) generateSymbiote (ExistsSymbiote SymbioteState{generate,generateOp,maxSize,generation}) = do let go g@SymbioteGeneration{size} = g {size = size + 1} @@ -155,8 +178,8 @@ generatedOperation <- encodeOp <$> genResize generateOp pure GeneratedSymbiote{generatedValue,generatedOperation} - +-- | The current \"progress\" or \"percent complete\" of the topic - the current 'size' divided by the 'maxSize'. getProgress :: MonadIO m => ExistsSymbiote s -> m Float getProgress (ExistsSymbiote SymbioteState{maxSize,generation}) = do - SymbioteGeneration{size} <- liftIO $ readTVarIO generation - pure $ fromIntegral size / fromIntegral maxSize + SymbioteGeneration{size} <- liftIO (readTVarIO generation) + pure (fromIntegral size / fromIntegral maxSize) diff --git a/src/Test/Serialization/Symbiote/Debug.hs b/src/Test/Serialization/Symbiote/Debug.hs --- a/src/Test/Serialization/Symbiote/Debug.hs +++ b/src/Test/Serialization/Symbiote/Debug.hs @@ -1,4 +1,22 @@ -module Test.Serialization.Symbiote.Debug where +{-| +Module: Test.Serialization.Symbiote.Debug +Copyright: (c) 2019 Athan Clark +License: BSD-3-Style +Maintainer: athan.clark@gmail.com +Portability: GHC +Data types that are used by the underlying protocol facilitating communication. See +'Test.Serialization.Symbiote.WebSocket' and 'Test.Serialization.Symbiote.ZeroMQ' for +details. + +-} + +module Test.Serialization.Symbiote.Debug where + +-- | Data type for deciding how verbose logging should be. data Debug = FullDebug | Percent | NoDebug + +-- | Data type determining if this is used on a public network or private one +data Network = Public | Private + deriving (Eq) diff --git a/src/Test/Serialization/Symbiote/WebSocket.hs b/src/Test/Serialization/Symbiote/WebSocket.hs --- a/src/Test/Serialization/Symbiote/WebSocket.hs +++ b/src/Test/Serialization/Symbiote/WebSocket.hs @@ -1,5 +1,6 @@ {-# LANGUAGE - RankNTypes + DataKinds + , RankNTypes , NamedFieldPuns , FlexibleContexts , ScopedTypeVariables @@ -22,9 +23,11 @@ import Test.Serialization.Symbiote (firstPeer, secondPeer, SymbioteT, defaultFailure, defaultProgress, nullProgress, Topic, Failure) -import Test.Serialization.Symbiote.Debug (Debug (..)) +import Test.Serialization.Symbiote.Debug (Debug (..), Network (..)) +import Test.Serialization.Symbiote.WebSocket.Ident (newWebSocketIdent, WithWebSocketIdent (..), WebSocketIdent) -import Data.Aeson (ToJSON, FromJSON, Value) +import Data.Aeson (ToJSON, FromJSON) +import qualified Data.Aeson as Json import Data.Serialize (Serialize) import qualified Data.ByteString as BS import qualified Data.ByteString.Lazy as LBS @@ -34,75 +37,81 @@ import Control.Monad.Trans.Control.Aligned (MonadBaseControl, liftBaseWith) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Catch (MonadCatch) +import Control.Concurrent (threadDelay) import Control.Concurrent.STM ( TChan, TMVar, newTChanIO, readTChan, writeTChan, atomically , newEmptyTMVarIO, putTMVar, takeTMVar) -import Control.Concurrent.Async (async, cancel, wait, Async) +import Control.Concurrent.STM.TChan.Typed (TChanRW, newTChanRW, writeTChanRW, readTChanRW) +import Control.Concurrent.Async (async, cancel, Async) +import Control.Concurrent.Chan.Scope (Scope (Read, Write)) +import Control.Concurrent.Chan.Extra (writeOnly) +import Control.Concurrent.Threaded.Hash (threaded) import Network.WebSockets.Simple (WebSocketsApp (..), WebSocketsAppParams (..), toClientAppTString, toClientAppTBinary, dimap', dimapJson, dimapStringify) import Network.WebSockets.Simple.Logger (logStdout) import Network.WebSockets.Trans (ClientAppT) +import System.Timeout (timeout) secondPeerWebSocketLazyByteString :: MonadIO m => MonadBaseControl IO m stM => MonadCatch m => Extractable stM - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => WebSocketParams -> Debug -> SymbioteT LBS.ByteString m () -- ^ Tests registered -> m () -secondPeerWebSocketLazyByteString host debug = peerWebSocketLazyByteString host debug secondPeer +secondPeerWebSocketLazyByteString params debug = peerWebSocketLazyByteString params debug secondPeer firstPeerWebSocketLazyByteString :: MonadIO m => MonadBaseControl IO m stM => MonadCatch m => Extractable stM - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => WebSocketParams -> Debug -> SymbioteT LBS.ByteString m () -- ^ Tests registered -> m () -firstPeerWebSocketLazyByteString host debug = peerWebSocketLazyByteString host debug firstPeer +firstPeerWebSocketLazyByteString params debug = peerWebSocketLazyByteString params debug firstPeer secondPeerWebSocketByteString :: MonadIO m => MonadBaseControl IO m stM => MonadCatch m => Extractable stM - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => WebSocketParams -> Debug -> SymbioteT BS.ByteString m () -- ^ Tests registered -> m () -secondPeerWebSocketByteString host debug = peerWebSocketByteString host debug secondPeer +secondPeerWebSocketByteString params debug = peerWebSocketByteString params debug secondPeer firstPeerWebSocketByteString :: MonadIO m => MonadBaseControl IO m stM => MonadCatch m => Extractable stM - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => WebSocketParams -> Debug -> SymbioteT BS.ByteString m () -- ^ Tests registered -> m () -firstPeerWebSocketByteString host debug = peerWebSocketByteString host debug firstPeer +firstPeerWebSocketByteString params debug = peerWebSocketByteString params debug firstPeer secondPeerWebSocketJson :: MonadIO m => MonadBaseControl IO m stM => MonadCatch m => Extractable stM - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => WebSocketParams -> Debug - -> SymbioteT Value m () -- ^ Tests registered + -> SymbioteT Json.Value m () -- ^ Tests registered -> m () -secondPeerWebSocketJson host debug = peerWebSocketJson host debug secondPeer +secondPeerWebSocketJson params debug = peerWebSocketJson params debug secondPeer firstPeerWebSocketJson :: MonadIO m => MonadBaseControl IO m stM => MonadCatch m => Extractable stM - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => WebSocketParams -> Debug - -> SymbioteT Value m () -- ^ Tests registered + -> SymbioteT Json.Value m () -- ^ Tests registered -> m () -firstPeerWebSocketJson host debug = peerWebSocketJson host debug firstPeer +firstPeerWebSocketJson params debug = peerWebSocketJson params debug firstPeer peerWebSocketLazyByteString :: forall m stM them me . MonadIO m @@ -112,7 +121,7 @@ => Show (them LBS.ByteString) => Serialize (me LBS.ByteString) => Serialize (them LBS.ByteString) - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => WebSocketParams -> Debug -> ( (me LBS.ByteString -> m ()) -> m (them LBS.ByteString) @@ -124,25 +133,132 @@ ) -> SymbioteT LBS.ByteString m () -- ^ Tests registered -> m () -peerWebSocketLazyByteString runClientAppT debug = peerWebSocket go debug - where - go :: WebSocketsApp IO (them LBS.ByteString) (me LBS.ByteString) -> IO () - go app = - runClientAppT $ toClientAppTBinary $ - ( case debug of - FullDebug -> logStdout - _ -> id - ) $ dimap' receive Cereal.encodeLazy app - where - receive :: LBS.ByteString -> IO (them LBS.ByteString) - receive buf = do - let eX = Cereal.decodeLazy buf - case eX of - Left e -> do - putStrLn $ "Can't parse buffer: " ++ show buf - error e - Right x -> pure x +peerWebSocketLazyByteString (WebSocketParams runWebSocket clientOrServer network) debug peer tests + | network == Private || (network == Public && clientOrServer == WebSocketClient) = do + (outgoing :: TChan (me LBS.ByteString)) <- liftIO newTChanIO + (incoming :: TChan (them LBS.ByteString)) <- liftIO newTChanIO + (outgoingThreadVar :: TMVar (Async ())) <- liftIO newEmptyTMVarIO + let encodeAndSend x = liftIO $ atomically $ writeTChan outgoing x + receiveAndDecode = liftIO $ atomically $ readTChan incoming + onSuccess t = liftIO $ putStrLn $ "WebSocket Topic finished: " ++ show t + onFailure = liftIO . defaultFailure + onProgress t n = case debug of + NoDebug -> nullProgress t n + _ -> liftIO (defaultProgress t n) + + webSocket :: forall a b. Serialize a => Serialize b => WebSocketsApp IO a b -> IO () + webSocket app = + runWebSocket $ toClientAppTBinary $ + ( case debug of + FullDebug -> logStdout + _ -> id + ) $ dimap' receive Cereal.encodeLazy app + where + receive :: LBS.ByteString -> IO a + receive buf = do + let eX = Cereal.decodeLazy buf + case eX of + Left e -> do + putStrLn ("Can't parse buffer: " ++ show buf) + error e + Right x -> pure x + wsThread <- case network of + Private -> do + let onOpen :: WebSocketsAppParams IO (me LBS.ByteString) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ atomically (readTChan outgoing) >>= send + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO (them LBS.ByteString) (me LBS.ByteString) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ -> atomically . writeTChan incoming + , onOpen + } + liftIO (async (webSocket app)) + Public -> do + ident <- liftIO newWebSocketIdent + let mkWsMessage = WithWebSocketIdent ident + + onOpen :: WebSocketsAppParams IO (WithWebSocketIdent (me LBS.ByteString)) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ do + x <- atomically (readTChan outgoing) + send (mkWsMessage x) + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO (WithWebSocketIdent (them LBS.ByteString)) (WithWebSocketIdent (me LBS.ByteString)) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ (WithWebSocketIdent _ x) -> atomically (writeTChan incoming x) + , onOpen + } + liftIO (async (webSocket app)) + + peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests + liftIO (cancel wsThread) + | otherwise = do + (incoming :: TChanRW 'Write (WebSocketIdent, them LBS.ByteString)) + <- writeOnly <$> liftIO (atomically newTChanRW) + + let process :: TChanRW 'Read (them LBS.ByteString) -> TChanRW 'Write (me LBS.ByteString) -> m () + process inputs outputs = void $ liftBaseWith $ \runInBase -> timeout 10000000 $ runInBase $ do + let encodeAndSend :: me LBS.ByteString -> m () + encodeAndSend x = liftIO $ atomically $ writeTChanRW outputs x + receiveAndDecode :: m (them LBS.ByteString) + receiveAndDecode = liftIO $ atomically $ readTChanRW inputs + + onSuccess t = liftIO $ putStrLn $ "WebSocket Topic finished: " ++ show t + onFailure = liftIO . defaultFailure + onProgress t n = case debug of + NoDebug -> nullProgress t n + _ -> liftIO (defaultProgress t n) + peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests + liftIO (threadDelay 1000000) + + ( _ + , outgoing :: TChanRW 'Read (WebSocketIdent, me LBS.ByteString) + ) <- threaded incoming process + (outgoingThreadVar :: TMVar (Async ())) <- liftIO newEmptyTMVarIO + + + let onOpen :: WebSocketsAppParams IO (WithWebSocketIdent (me LBS.ByteString)) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ do + (ident, x) <- atomically (readTChanRW outgoing) + send (WithWebSocketIdent ident x) + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO + (WithWebSocketIdent (them LBS.ByteString)) + (WithWebSocketIdent (me LBS.ByteString)) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ (WithWebSocketIdent ident x) -> atomically (writeTChanRW incoming (ident, x)) + , onOpen + } + webSocket :: IO () + webSocket = + runWebSocket $ toClientAppTBinary $ + ( case debug of + FullDebug -> logStdout + _ -> id + ) $ dimap' receive Cereal.encodeLazy app + where + receive :: LBS.ByteString -> IO (WithWebSocketIdent (them LBS.ByteString)) + receive buf = do + let eX = Cereal.decodeLazy buf + case eX of + Left e -> do + putStrLn ("Can't parse buffer: " ++ show buf) + error e + Right x -> pure x + liftIO webSocket + peerWebSocketByteString :: forall m stM them me . MonadIO m => MonadBaseControl IO m stM @@ -151,7 +267,7 @@ => Show (them BS.ByteString) => Serialize (me BS.ByteString) => Serialize (them BS.ByteString) - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => WebSocketParams -> Debug -> ( (me BS.ByteString -> m ()) -> m (them BS.ByteString) @@ -160,109 +276,297 @@ -> (Topic -> Float -> m ()) -> SymbioteT BS.ByteString m () -> m () - ) + ) -- ^ Encode and send, receive and decode, on success, on failure, on progress, and test set -> SymbioteT BS.ByteString m () -- ^ Tests registered -> m () -peerWebSocketByteString runClientAppT debug = peerWebSocket go debug - where - go :: WebSocketsApp IO (them BS.ByteString) (me BS.ByteString) -> IO () - go app = - runClientAppT $ toClientAppTBinary $ toLazy $ - ( case debug of - FullDebug -> logStdout - _ -> id - ) $ dimap' receive Cereal.encode app - where - receive :: BS.ByteString -> IO (them BS.ByteString) - receive buf = do - let eX = Cereal.decode buf - case eX of - Left e -> do - putStrLn $ "Can't parse buffer: " ++ show buf - error e - Right x -> pure x +peerWebSocketByteString (WebSocketParams runWebSocket clientOrServer network) debug peer tests + | network == Private || (network == Public && clientOrServer == WebSocketClient) = do + (outgoing :: TChan (me BS.ByteString)) <- liftIO newTChanIO + (incoming :: TChan (them BS.ByteString)) <- liftIO newTChanIO + (outgoingThreadVar :: TMVar (Async ())) <- liftIO newEmptyTMVarIO - toLazy :: WebSocketsApp IO BS.ByteString BS.ByteString -> WebSocketsApp IO LBS.ByteString LBS.ByteString - toLazy = dimap' r s - where - r = pure . LBS.toStrict - s = LBS.fromStrict + let encodeAndSend x = liftIO $ atomically $ writeTChan outgoing x + receiveAndDecode = liftIO $ atomically $ readTChan incoming + onSuccess t = liftIO $ putStrLn $ "WebSocket Topic finished: " ++ show t + onFailure = liftIO . defaultFailure + onProgress t n = case debug of + NoDebug -> nullProgress t n + _ -> liftIO (defaultProgress t n) + webSocket :: forall a b. Serialize a => Serialize b => WebSocketsApp IO a b -> IO () + webSocket app = + runWebSocket $ toClientAppTBinary $ toLazy $ + ( case debug of + FullDebug -> logStdout + _ -> id + ) $ dimap' receive Cereal.encode app + where + receive :: BS.ByteString -> IO a + receive buf = do + let eX = Cereal.decode buf + case eX of + Left e -> do + putStrLn ("Can't parse buffer: " ++ show buf) + error e + Right x -> pure x -peerWebSocketJson :: MonadIO m + toLazy :: WebSocketsApp IO BS.ByteString BS.ByteString + -> WebSocketsApp IO LBS.ByteString LBS.ByteString + toLazy = dimap' r s + where + r = pure . LBS.toStrict + s = LBS.fromStrict + wsThread <- case network of + Private -> do + let onOpen :: WebSocketsAppParams IO (me BS.ByteString) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ atomically (readTChan outgoing) >>= send + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO (them BS.ByteString) (me BS.ByteString) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ -> atomically . writeTChan incoming + , onOpen + } + liftIO (async (webSocket app)) + Public -> do + ident <- liftIO newWebSocketIdent + let mkWsMessage = WithWebSocketIdent ident + + onOpen :: WebSocketsAppParams IO (WithWebSocketIdent (me BS.ByteString)) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ do + x <- atomically (readTChan outgoing) + send (mkWsMessage x) + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO (WithWebSocketIdent (them BS.ByteString)) (WithWebSocketIdent (me BS.ByteString)) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ (WithWebSocketIdent _ x) -> atomically (writeTChan incoming x) + , onOpen + } + liftIO (async (webSocket app)) + + peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests + liftIO (cancel wsThread) + | otherwise = do + (incoming :: TChanRW 'Write (WebSocketIdent, them BS.ByteString)) + <- writeOnly <$> liftIO (atomically newTChanRW) + + let process :: TChanRW 'Read (them BS.ByteString) -> TChanRW 'Write (me BS.ByteString) -> m () + process inputs outputs = void $ liftBaseWith $ \runInBase -> timeout 10000000 $ runInBase $ do + let encodeAndSend :: me BS.ByteString -> m () + encodeAndSend x = liftIO $ atomically $ writeTChanRW outputs x + receiveAndDecode :: m (them BS.ByteString) + receiveAndDecode = liftIO $ atomically $ readTChanRW inputs + + onSuccess t = liftIO $ putStrLn $ "WebSocket Topic finished: " ++ show t + onFailure = liftIO . defaultFailure + onProgress t n = case debug of + NoDebug -> nullProgress t n + _ -> liftIO (defaultProgress t n) + peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests + liftIO (threadDelay 1000000) + + ( _ + , outgoing :: TChanRW 'Read (WebSocketIdent, me BS.ByteString) + ) <- threaded incoming process + (outgoingThreadVar :: TMVar (Async ())) <- liftIO newEmptyTMVarIO + + + let onOpen :: WebSocketsAppParams IO (WithWebSocketIdent (me BS.ByteString)) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ do + (ident, x) <- atomically (readTChanRW outgoing) + send (WithWebSocketIdent ident x) + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO + (WithWebSocketIdent (them BS.ByteString)) + (WithWebSocketIdent (me BS.ByteString)) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ (WithWebSocketIdent ident x) -> atomically (writeTChanRW incoming (ident, x)) + , onOpen + } + webSocket :: IO () + webSocket = + runWebSocket $ toClientAppTBinary $ toLazy $ + ( case debug of + FullDebug -> logStdout + _ -> id + ) $ dimap' receive Cereal.encode app + where + receive :: BS.ByteString -> IO (WithWebSocketIdent (them BS.ByteString)) + receive buf = do + let eX = Cereal.decode buf + case eX of + Left e -> do + putStrLn ("Can't parse buffer: " ++ show buf) + error e + Right x -> pure x + + toLazy :: WebSocketsApp IO BS.ByteString BS.ByteString + -> WebSocketsApp IO LBS.ByteString LBS.ByteString + toLazy = dimap' r s + where + r = pure . LBS.toStrict + s = LBS.fromStrict + liftIO webSocket + + +-- WebSockets can work with both Json 'Value's and 'BS.ByteString's +peerWebSocketJson :: forall m stM them me + . MonadIO m => MonadBaseControl IO m stM => Extractable stM - => MonadCatch m - => Show (them Value) - => ToJSON (me Value) - => FromJSON (them Value) - => (ClientAppT IO () -> IO ()) -- ^ Run the generated WebSocket client + => Show (them Json.Value) + => ToJSON (me Json.Value) + => FromJSON (them Json.Value) + => WebSocketParams -> Debug - -> ( (me Value -> m ()) - -> m (them Value) + -> ( (me Json.Value -> m ()) + -> m (them Json.Value) -> (Topic -> m ()) - -> (Failure them Value -> m ()) + -> (Failure them Json.Value -> m ()) -> (Topic -> Float -> m ()) - -> SymbioteT Value m () + -> SymbioteT Json.Value m () -> m () - ) - -> SymbioteT Value m () -- ^ Tests registered + ) -- ^ Encode and send, receive and decode, on success, on failure, on progress, and test set + -> SymbioteT Json.Value m () -- ^ Tests registered -> m () -peerWebSocketJson runClientAppT debug = peerWebSocket - ( runClientAppT - . toClientAppTString - . ( case debug of - FullDebug -> logStdout - _ -> id - ) - . dimapStringify - . dimapJson - ) - debug +peerWebSocketJson (WebSocketParams runWebSocket clientOrServer network) debug peer tests + | network == Private || (network == Public && clientOrServer == WebSocketClient) = do + (outgoing :: TChan (me Json.Value)) <- liftIO newTChanIO + (incoming :: TChan (them Json.Value)) <- liftIO newTChanIO + (outgoingThreadVar :: TMVar (Async ())) <- liftIO newEmptyTMVarIO -peerWebSocket :: forall m stM s them me - . MonadIO m - => MonadBaseControl IO m stM - => Show (them s) - => Show s - => ( WebSocketsApp IO (them s) (me s) - -> IO () - ) -- ^ Run the generated WebSocketsApp - -> Debug - -> ( (me s -> m ()) - -> m (them s) - -> (Topic -> m ()) - -> (Failure them s -> m ()) - -> (Topic -> Float -> m ()) - -> SymbioteT s m () - -> m () - ) - -> SymbioteT s m () -- ^ Tests registered - -> m () -peerWebSocket webSocket debug peer tests = do - (outgoing :: TChan (me s)) <- liftIO newTChanIO - (incoming :: TChan (them s)) <- liftIO newTChanIO - (outgoingThreadVar :: TMVar (Async ())) <- liftIO newEmptyTMVarIO - let encodeAndSend x = liftIO $ atomically $ writeTChan outgoing x - receiveAndDecode = liftIO $ atomically $ readTChan incoming - onSuccess t = liftIO $ putStrLn $ "Topic finished: " ++ show t - onFailure = liftIO . defaultFailure - onProgress t n = case debug of - NoDebug -> nullProgress t n - _ -> liftIO (defaultProgress t n) + let encodeAndSend x = liftIO $ atomically $ writeTChan outgoing x + receiveAndDecode = liftIO $ atomically $ readTChan incoming + onSuccess t = liftIO $ putStrLn $ "WebSocket Topic finished: " ++ show t + onFailure = liftIO . defaultFailure + onProgress t n = case debug of + NoDebug -> nullProgress t n + _ -> liftIO (defaultProgress t n) + webSocket :: forall a b. FromJSON a => ToJSON b => WebSocketsApp IO a b -> IO () + webSocket = + runWebSocket + . toClientAppTString + . ( case debug of + FullDebug -> logStdout + _ -> id + ) + . dimapStringify + . dimapJson - onOpen :: WebSocketsAppParams IO (me s) -> IO () - onOpen WebSocketsAppParams{close,send} = do - outgoingThread <- async $ forever $ atomically (readTChan outgoing) >>= send - atomically (putTMVar outgoingThreadVar outgoingThread) - app :: WebSocketsApp IO (them s) (me s) - app = WebSocketsApp - { onClose = \_ _ -> do - outgoingThread <- atomically (takeTMVar outgoingThreadVar) - cancel outgoingThread - , onReceive = \_ -> atomically . writeTChan incoming - , onOpen - } - wsThread <- liftIO (async (webSocket app)) - peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests - liftIO (cancel wsThread) + wsThread <- case network of + Private -> do + let onOpen :: WebSocketsAppParams IO (me Json.Value) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ atomically (readTChan outgoing) >>= send + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO (them Json.Value) (me Json.Value) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ -> atomically . writeTChan incoming + , onOpen + } + + liftIO (async (webSocket app)) + Public -> do + ident <- liftIO newWebSocketIdent + let mkWsMessage = WithWebSocketIdent ident + + onOpen :: WebSocketsAppParams IO (WithWebSocketIdent (me Json.Value)) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ do + x <- atomically (readTChan outgoing) + send (mkWsMessage x) + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO (WithWebSocketIdent (them Json.Value)) (WithWebSocketIdent (me Json.Value)) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ (WithWebSocketIdent _ x) -> atomically (writeTChan incoming x) + , onOpen + } + liftIO (async (webSocket app)) + + peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests + liftIO (cancel wsThread) + | otherwise = do + (incoming :: TChanRW 'Write (WebSocketIdent, them Json.Value)) + <- writeOnly <$> liftIO (atomically newTChanRW) + + let process :: TChanRW 'Read (them Json.Value) -> TChanRW 'Write (me Json.Value) -> m () + process inputs outputs = void $ liftBaseWith $ \runInBase -> timeout 10000000 $ runInBase $ do + let encodeAndSend :: me Json.Value -> m () + encodeAndSend x = liftIO $ atomically $ writeTChanRW outputs x + receiveAndDecode :: m (them Json.Value) + receiveAndDecode = liftIO $ atomically $ readTChanRW inputs + + onSuccess t = liftIO $ putStrLn $ "WebSocket Topic finished: " ++ show t + onFailure = liftIO . defaultFailure + onProgress t n = case debug of + NoDebug -> nullProgress t n + _ -> liftIO (defaultProgress t n) + peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests + liftIO (threadDelay 1000000) + + ( _ + , outgoing :: TChanRW 'Read (WebSocketIdent, me Json.Value) + ) <- threaded incoming process + (outgoingThreadVar :: TMVar (Async ())) <- liftIO newEmptyTMVarIO + + + let onOpen :: WebSocketsAppParams IO (WithWebSocketIdent (me Json.Value)) -> IO () + onOpen WebSocketsAppParams{send} = do + outgoingThread <- async $ forever $ do + (ident, x) <- atomically (readTChanRW outgoing) + send (WithWebSocketIdent ident x) + atomically (putTMVar outgoingThreadVar outgoingThread) + app :: WebSocketsApp IO + (WithWebSocketIdent (them Json.Value)) + (WithWebSocketIdent (me Json.Value)) + app = WebSocketsApp + { onClose = \_ _ -> do + outgoingThread <- atomically (takeTMVar outgoingThreadVar) + cancel outgoingThread + , onReceive = \_ (WithWebSocketIdent ident x) -> atomically (writeTChanRW incoming (ident, x)) + , onOpen + } + webSocket :: WebSocketsApp IO + (WithWebSocketIdent (them Json.Value)) + (WithWebSocketIdent (me Json.Value)) + -> IO () + webSocket = + runWebSocket + . toClientAppTString + . ( case debug of + FullDebug -> logStdout + _ -> id + ) + . dimapStringify + . dimapJson + liftIO (webSocket app) + + + +data WebSocketServerOrClient + = WebSocketServer + | WebSocketClient + deriving (Eq) + +data WebSocketParams = WebSocketParams + { runWebSocket :: ClientAppT IO () -> IO () -- ^ Run the generated WebSocketsApp + , webSocketServerOrClient :: WebSocketServerOrClient + , webSocketNetwork :: Network + } diff --git a/src/Test/Serialization/Symbiote/WebSocket/Ident.hs b/src/Test/Serialization/Symbiote/WebSocket/Ident.hs new file mode 100644 --- /dev/null +++ b/src/Test/Serialization/Symbiote/WebSocket/Ident.hs @@ -0,0 +1,85 @@ +{-# LANGUAGE + GeneralizedNewtypeDeriving + , RecordWildCards + , NamedFieldPuns + , OverloadedStrings + #-} + +{-| + +Module: Test.Serialization.Symbiote.WebSocket.Ident +Copyright: (c) 2019 Athan Clark +License: BSD-3-Style +Maintainer: athan.clark@gmail.com +Portability: GHC + +Data types used for WebSocket implementations + +-} + +module Test.Serialization.Symbiote.WebSocket.Ident where + +import Data.UUID (UUID, toText, fromText, toWords, fromWords) +import Data.UUID.V4 (nextRandom) +import Data.Aeson (FromJSON (..), ToJSON (..), Value (String, Object), object, (.:), (.=)) +import Data.Aeson.Types (typeMismatch) +import Data.Serialize (Serialize (..)) +import Data.Serialize.Get (getWord32be) +import Data.Serialize.Put (putWord32be) +import Data.Hashable (Hashable) +import Test.QuickCheck (Arbitrary (..)) +import System.IO.Unsafe (unsafePerformIO) + + +newtype WebSocketIdent = WebSocketIdent {getWebSocketIdent :: UUID} + deriving (Show, Eq, Ord, Hashable) + +newWebSocketIdent :: IO WebSocketIdent +newWebSocketIdent = WebSocketIdent <$> nextRandom + +instance ToJSON WebSocketIdent where + toJSON (WebSocketIdent x) = String (toText x) + +instance FromJSON WebSocketIdent where + parseJSON json = case json of + String s -> case fromText s of + Nothing -> fail' + Just x -> pure (WebSocketIdent x) + _ -> fail' + where + fail' = typeMismatch "WebSocketIdent" json + +instance Serialize WebSocketIdent where + get = WebSocketIdent <$> (fromWords <$> getWord32be <*> getWord32be <*> getWord32be <*> getWord32be) + put (WebSocketIdent x) = + let (a,b,c,d) = toWords x + in putWord32be a *> putWord32be b *> putWord32be c *> putWord32be d + +instance Arbitrary WebSocketIdent where + arbitrary = pure (unsafePerformIO newWebSocketIdent) + + +data WithWebSocketIdent a = WithWebSocketIdent + { webSocketIdent :: WebSocketIdent + , webSocketValue :: a + } deriving (Show, Eq) + +instance ToJSON a => ToJSON (WithWebSocketIdent a) where + toJSON WithWebSocketIdent{..} = object ["ident" .= webSocketIdent, "value" .= webSocketValue] + +instance FromJSON a => FromJSON (WithWebSocketIdent a) where + parseJSON json = case json of + Object o -> WithWebSocketIdent <$> o .: "ident" <*> o .: "value" + _ -> typeMismatch "WithWebSocketIdent" json + +instance Serialize a => Serialize (WithWebSocketIdent a) where + get = do + webSocketIdent <- get + webSocketValue <- get + pure WithWebSocketIdent{webSocketIdent,webSocketValue} + put WithWebSocketIdent{..} = do + put webSocketIdent + put webSocketValue + +instance Arbitrary a => Arbitrary (WithWebSocketIdent a) where + arbitrary = WithWebSocketIdent <$> arbitrary <*> arbitrary diff --git a/src/Test/Serialization/Symbiote/ZeroMQ.hs b/src/Test/Serialization/Symbiote/ZeroMQ.hs --- a/src/Test/Serialization/Symbiote/ZeroMQ.hs +++ b/src/Test/Serialization/Symbiote/ZeroMQ.hs @@ -1,5 +1,6 @@ {-# LANGUAGE - RankNTypes + DataKinds + , RankNTypes , FlexibleContexts , ScopedTypeVariables #-} @@ -21,51 +22,68 @@ import Test.Serialization.Symbiote (firstPeer, secondPeer, SymbioteT, defaultFailure, defaultProgress, nullProgress, Topic, Failure) -import Test.Serialization.Symbiote.Debug (Debug (..)) +import Test.Serialization.Symbiote.Debug (Debug (..), Network (..)) import qualified Data.ByteString as BS import qualified Data.Serialize as Cereal import Data.List.NonEmpty (NonEmpty (..)) import Data.Singleton.Class (Extractable) -import Control.Monad (forever) -import Control.Monad.Catch (MonadCatch) +import Control.Monad (forever, void) import Control.Monad.IO.Class (MonadIO (liftIO)) -import Control.Monad.Trans.Control.Aligned (MonadBaseControl) -import Control.Concurrent.Async (Async, cancel) -import Control.Concurrent.STM (TChan, TMVar, newTChanIO, writeTChan, readTChan, newEmptyTMVarIO, putTMVar, takeTMVar, atomically) -import System.ZMQ4 (Pair (..)) +import Control.Monad.Trans.Control.Aligned (MonadBaseControl, liftBaseWith) +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async (cancel) +import Control.Concurrent.Chan.Scope (Scope (Read, Write)) +import Control.Concurrent.Chan.Extra (writeOnly) +import Control.Concurrent.STM (TChan, newTChanIO, writeTChan, readTChan, atomically) +import Control.Concurrent.STM.TChan.Typed (TChanRW, newTChanRW, writeTChanRW, readTChanRW) +import Control.Concurrent.Threaded.Hash (threaded) +import System.ZMQ4 (Router (..), Dealer (..), Pair (..)) import System.ZMQ4.Monadic (runZMQ, async) -import System.ZMQ4.Simple (socket, bind, send, receive) +import System.ZMQ4.Simple (ZMQIdent, socket, bind, send, receive, connect, setUUIDIdentity) +import System.Timeout (timeout) secondPeerZeroMQ :: MonadIO m => MonadBaseControl IO m stM - => MonadCatch m => Extractable stM - => String + => ZeroMQParams -> Debug -> SymbioteT BS.ByteString m () -- ^ Tests registered -> m () -secondPeerZeroMQ host debug = peerZeroMQ host debug secondPeer +secondPeerZeroMQ params debug = peerZeroMQ params debug secondPeer firstPeerZeroMQ :: MonadIO m => MonadBaseControl IO m stM - => MonadCatch m => Extractable stM - => String + => ZeroMQParams -> Debug -> SymbioteT BS.ByteString m () -- ^ Tests registered -> m () -firstPeerZeroMQ host debug = peerZeroMQ host debug firstPeer +firstPeerZeroMQ params debug = peerZeroMQ params debug firstPeer + +data ZeroMQServerOrClient + = ZeroMQServer + | ZeroMQClient + +data ZeroMQParams = ZeroMQParams + { zmqHost :: String + , zmqServerOrClient :: ZeroMQServerOrClient + , zmqNetwork :: Network + } + + +-- | ZeroMQ can only work on 'BS.ByteString's peerZeroMQ :: forall m stM them me . MonadIO m => MonadBaseControl IO m stM + => Extractable stM => Show (them BS.ByteString) => Cereal.Serialize (me BS.ByteString) => Cereal.Serialize (them BS.ByteString) - => String -- ^ Host + => ZeroMQParams -> Debug -> ( (me BS.ByteString -> m ()) -> m (them BS.ByteString) @@ -74,33 +92,116 @@ -> (Topic -> Float -> m ()) -> SymbioteT BS.ByteString m () -> m () - ) + ) -- ^ Encode and send, receive and decode, on success, on failure, on progress, and test set -> SymbioteT BS.ByteString m () -- ^ Tests registered -> m () -peerZeroMQ host debug peer tests = do - (outgoing :: TChan (me BS.ByteString)) <- liftIO newTChanIO - (incoming :: TChan (them BS.ByteString)) <- liftIO newTChanIO - (outgoingThreadVar :: TMVar (Async ())) <- liftIO newEmptyTMVarIO - let encodeAndSend x = liftIO $ atomically $ writeTChan outgoing x - receiveAndDecode = liftIO $ atomically $ readTChan incoming - onSuccess t = liftIO $ putStrLn $ "Topic finished: " ++ show t - onFailure = liftIO . defaultFailure - onProgress t n = case debug of - NoDebug -> nullProgress t n - _ -> liftIO (defaultProgress t n) - zThread <- runZMQ $ async $ do - s <- socket Pair Pair - bind s host - outgoingThread <- async $ forever $ do - x <- liftIO (atomically (readTChan outgoing)) - send () s ((Cereal.encode x) :| []) - liftIO (atomically (putTMVar outgoingThreadVar outgoingThread)) - forever $ do - mX <- receive s - case mX of - Nothing -> liftIO $ putStrLn "got nothin" - Just ((),x :| _) -> case Cereal.decode x of - Left e -> error $ "couldn't decode: " ++ e - Right x' -> liftIO (atomically (writeTChan incoming x')) - peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests - liftIO (cancel zThread) +peerZeroMQ (ZeroMQParams host clientOrServer network) debug peer tests = + case (network,clientOrServer) of + (Public,ZeroMQServer) -> do + (incoming :: TChanRW 'Write (ZMQIdent, them BS.ByteString)) <- writeOnly <$> liftIO (atomically newTChanRW) + -- the process that gets invoked for each new thread. Writes to a @me BS.ByteString@ and reads from a @them BS.ByteString@. + let process :: TChanRW 'Read (them BS.ByteString) -> TChanRW 'Write (me BS.ByteString) -> m () + process inputs outputs = void $ liftBaseWith $ \runInBase -> timeout 10000000 $ runInBase $ do + let encodeAndSend :: me BS.ByteString -> m () + encodeAndSend x = liftIO $ atomically $ writeTChanRW outputs x + + receiveAndDecode :: m (them BS.ByteString) + receiveAndDecode = liftIO $ atomically $ readTChanRW inputs + + onSuccess t = liftIO $ putStrLn $ "ZeroMQ Topic finished: " ++ show t + onFailure = liftIO . defaultFailure + onProgress t n = case debug of + NoDebug -> nullProgress t n + _ -> liftIO (defaultProgress t n) + peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests + liftIO (threadDelay 1000000) + + -- manage invoked threads + ( _ + , outgoing :: TChanRW 'Read (ZMQIdent, me BS.ByteString) + ) <- threaded incoming process + + -- forever bind to ZeroMQ + runZMQ $ do + s <- socket Router Dealer + bind s host + + -- sending loop (separate thread) + void $ async $ forever $ do + (ident, x) <- liftIO (atomically (readTChanRW outgoing)) + send ident s ((Cereal.encode x) :| []) + + -- receiving loop (current thread) + forever $ do + mX <- receive s + case mX of + Nothing -> liftIO $ putStrLn "got nothin" + Just (ident,x :| _) -> case Cereal.decode x of + Left e -> error $ "couldn't decode: " ++ e + Right x' -> do + liftIO (atomically (writeTChanRW incoming (ident,x'))) + -- liftIO (cancel zThread) - never dies automatically? + + _ -> do + (outgoing :: TChan (me BS.ByteString)) <- liftIO newTChanIO + (incoming :: TChan (them BS.ByteString)) <- liftIO newTChanIO + let encodeAndSend :: me BS.ByteString -> m () + encodeAndSend x = liftIO (atomically (writeTChan outgoing x)) + + receiveAndDecode :: m (them BS.ByteString) + receiveAndDecode = liftIO (atomically (readTChan incoming)) + + onSuccess t = liftIO $ putStrLn $ "ZeroMQ Topic finished: " ++ show t + onFailure = liftIO . defaultFailure + onProgress t n = case debug of + NoDebug -> nullProgress t n + _ -> liftIO (defaultProgress t n) + + -- sending loop (separate thread) + sendingThread s = + void $ async $ forever $ do + x <- liftIO (atomically (readTChan outgoing)) + send () s ((Cereal.encode x) :| []) + + -- receiving loop (current thread) + receivingLoop s = forever $ do + mX <- receive s + case mX of + Nothing -> liftIO (putStrLn "got nothin") + Just ((),x :| _) -> case Cereal.decode x of + Left e -> error ("couldn't decode: " ++ e) + Right x' -> do + liftIO (atomically (writeTChan incoming x')) + + -- thread that connects and communicates with ZeroMQ + zThread <- case network of + -- is a ZeroMQClient + Public -> runZMQ $ async $ do + s <- socket Dealer Router + setUUIDIdentity s + connect s host + + sendingThread s + + receivingLoop s + Private -> case clientOrServer of + ZeroMQServer -> runZMQ $ async $ do + s <- socket Pair Pair + bind s host + + sendingThread s + + receivingLoop s + ZeroMQClient -> runZMQ $ async $ do + s <- socket Pair Pair + connect s host + + sendingThread s + + receivingLoop s + + -- main loop (current thread, continues when finished) + peer encodeAndSend receiveAndDecode onSuccess onFailure onProgress tests + + -- kill ZeroMQ thread + liftIO (cancel zThread) diff --git a/symbiote.cabal b/symbiote.cabal --- a/symbiote.cabal +++ b/symbiote.cabal @@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack -- --- hash: b6ff1fa8d8f7f7eb06962e6726c32348cec4ff8a0d70f61b0628093fbdc88029 +-- hash: bb816980e733fd1d164875e94302b3003d655cc593b1de79483b12e50a73761b name: symbiote -version: 0.0.2 +version: 0.0.3 synopsis: Data serialization, communication, and operation verification implementation description: Please see the README on GitHub at category: Data, Testing @@ -15,7 +15,7 @@ bug-reports: https://github.com/athanclark/symbiote/issues author: Athan Clark maintainer: athan.clark@gmail.com -copyright: 2019 Athan Clark +copyright: 2019, 2020 Athan Clark license: BSD3 license-file: LICENSE build-type: Simple @@ -37,6 +37,7 @@ Test.Serialization.Symbiote.Core Test.Serialization.Symbiote.Debug Test.Serialization.Symbiote.WebSocket + Test.Serialization.Symbiote.WebSocket.Ident Test.Serialization.Symbiote.ZeroMQ other-modules: Paths_symbiote @@ -50,14 +51,18 @@ , base >=4.7 && <5 , bytestring , cereal + , chan , containers , exceptions , extractable-singleton + , hashable , monad-control-aligned , mtl , quickcheck-instances , stm , text + , threaded + , uuid , wai-transformers >=0.1.0 , websockets-simple >=0.2.0 , websockets-simple-extra @@ -69,6 +74,7 @@ type: exitcode-stdio-1.0 main-is: Spec.hs other-modules: + ServerOrClient Spec.Local Spec.Protocol Spec.Sanity @@ -85,9 +91,11 @@ , base >=4.7 && <5 , bytestring , cereal + , chan , containers , exceptions , extractable-singleton + , hashable , http-types , monad-control-aligned , mtl @@ -98,6 +106,8 @@ , tasty-hunit , tasty-quickcheck , text + , threaded + , uuid , wai , wai-extra , wai-transformers diff --git a/test/ServerOrClient.hs b/test/ServerOrClient.hs new file mode 100644 --- /dev/null +++ b/test/ServerOrClient.hs @@ -0,0 +1,26 @@ +{-# LANGUAGE + OverloadedStrings + #-} + +module ServerOrClient where + +import Test.Tasty.Options (safeReadBool, flagCLParser, IsOption (..)) +import Test.Serialization.Symbiote.Debug (Network (..)) +import Data.Bool (bool) + + +data ServerOrClient = Server | Client + +instance IsOption ServerOrClient where + defaultValue = Server + parseValue s = (bool Server Client) <$> safeReadBool s + optionName = "is-client" + optionHelp = "Set to 'true' when you want the test suite to behave as a client" + optionCLParser = flagCLParser (Just 'c') Client + + +instance IsOption Network where + defaultValue = Private + parseValue s = (bool Private Public) <$> safeReadBool s + optionName = "is-public" + optionHelp = "Set to 'true' when you want the test suite to behave on a public channel" diff --git a/test/Spec.hs b/test/Spec.hs --- a/test/Spec.hs +++ b/test/Spec.hs @@ -10,14 +10,22 @@ import Spec.Sanity (sanityTests) import Spec.Local (localIsos) import Spec.Protocol (protocolTests) - -import Test.Tasty (defaultMain, testGroup, TestTree) +import Test.Serialization.Symbiote.Debug (Network) +import Test.Tasty (defaultMainWithIngredients, defaultIngredients, includingOptions, testGroup, TestTree) +import Test.Tasty.Options (OptionDescription (..)) +import ServerOrClient (ServerOrClient) +import Data.Proxy (Proxy (..)) main :: IO () -main = defaultMain tests - +main = + defaultMainWithIngredients + (includingOptions [serverOrClientArg, networkArg] : defaultIngredients) + tests + where + serverOrClientArg = Option (Proxy :: Proxy ServerOrClient) + networkArg = Option (Proxy :: Proxy Network) tests :: TestTree tests = testGroup "All Tests" $ diff --git a/test/Spec/Local.hs b/test/Spec/Local.hs --- a/test/Spec/Local.hs +++ b/test/Spec/Local.hs @@ -11,6 +11,7 @@ import Test.QuickCheck.Instances () import Test.Serialization.Symbiote (First, Second, Generating, Operating, Topic) import Test.Serialization.Symbiote.Abides +import Test.Serialization.Symbiote.WebSocket.Ident (WebSocketIdent, WithWebSocketIdent) localIsos :: TestTree @@ -37,6 +38,8 @@ , go ("First Int", Proxy :: Proxy (First Int)) , go ("Second Int", Proxy :: Proxy (Second Int)) , go ("Topic", Proxy :: Proxy Topic) + , go ("WebSocketIdent", Proxy :: Proxy WebSocketIdent) + , go ("WithWebSocketIdent", Proxy :: Proxy (WithWebSocketIdent Int)) ] ] , testGroup "Cereal" $ @@ -60,6 +63,8 @@ , go ("First ByteString", Proxy :: Proxy (First ByteString)) , go ("Second ByteString", Proxy :: Proxy (Second ByteString)) , go ("Topic", Proxy :: Proxy Topic) + , go ("WebSocketIdent", Proxy :: Proxy WebSocketIdent) + , go ("WithWebSocketIdent", Proxy :: Proxy (WithWebSocketIdent Int)) ] ] ] diff --git a/test/Spec/Protocol.hs b/test/Spec/Protocol.hs --- a/test/Spec/Protocol.hs +++ b/test/Spec/Protocol.hs @@ -8,24 +8,28 @@ module Spec.Protocol where +import ServerOrClient (ServerOrClient (..)) + import qualified Data.Aeson as Json import qualified Data.Aeson.Types as Json import qualified Data.Serialize as Cereal -import qualified Data.Serialize.Get as Cereal -import qualified Data.Serialize.Put as Cereal import qualified Data.ByteString as BS import Data.Proxy -import Control.Monad.IO.Class (liftIO) import Test.Serialization.Symbiote (SymbioteOperation (..), Generating, Operating, First, Second, Topic, SymbioteT, register) import Test.Serialization.Symbiote.Debug (Debug (..)) -import Test.Serialization.Symbiote.WebSocket (firstPeerWebSocketJson, firstPeerWebSocketByteString) -import Test.Serialization.Symbiote.ZeroMQ (firstPeerZeroMQ) +import Test.Serialization.Symbiote.WebSocket + ( firstPeerWebSocketJson, firstPeerWebSocketByteString + , secondPeerWebSocketJson, secondPeerWebSocketByteString + , WebSocketParams (..), WebSocketServerOrClient (..)) +import Test.Serialization.Symbiote.ZeroMQ + ( firstPeerZeroMQ, secondPeerZeroMQ, ZeroMQParams (..), ZeroMQServerOrClient (..)) import Test.Serialization.Symbiote.Aeson () import Test.Serialization.Symbiote.Cereal () import Test.QuickCheck (Arbitrary (..)) -import Test.Tasty (TestTree, testGroup) +import Test.Tasty (TestTree, testGroup, askOption) import Test.Tasty.HUnit (testCase) +import Network.WebSockets (runClient) import Network.WebSockets.Connection (defaultConnectionOptions) import Network.WebSockets.Simple (accept) import Network.WebSockets.Trans (runServerAppT) @@ -38,50 +42,90 @@ protocolTests :: [TestTree] protocolTests = - [ testGroup "WebSocket Server" - [ testCase "Json" $ - let tests :: SymbioteT Json.Value IO () - tests = do - register "Generating Topic" 100 (Proxy :: Proxy (Generating Topic)) - register "Operating Topic" 100 (Proxy :: Proxy (Operating Topic)) - register "First Topic" 100 (Proxy :: Proxy (First Topic)) - register "Second Topic" 100 (Proxy :: Proxy (Second Topic)) - register "Topic" 100 (Proxy :: Proxy Topic) - runClient client = do - -- runServer "localhost" 3000 server' - let server = accept client - server' <- runServerAppT server - liftIO $ run 3000 $ logStdoutDev $ websocketsOr defaultConnectionOptions server' $ \_ respond -> - respond $ responseLBS status400 [] "Not a websocket" - in firstPeerWebSocketJson runClient NoDebug tests - , testCase "ByteString" $ - let tests :: SymbioteT BS.ByteString IO () - tests = do - register "Generating ByteString" 50 (Proxy :: Proxy (Generating BS.ByteString)) - register "Operating ByteString" 50 (Proxy :: Proxy (Operating BS.ByteString)) - register "First ByteString" 50 (Proxy :: Proxy (First BS.ByteString)) - register "Second ByteString" 50 (Proxy :: Proxy (Second BS.ByteString)) - register "Topic" 50 (Proxy :: Proxy Topic) - runClient client = do - -- runServer "localhost" 3000 server' - let server = accept client - server' <- runServerAppT server - liftIO $ run 3001 $ logStdoutDev $ websocketsOr defaultConnectionOptions server' $ \_ respond -> - respond $ responseLBS status400 [] "Not a websocket" - in firstPeerWebSocketByteString runClient NoDebug tests - ] - , testGroup "ZeroMQ Server" - [ testCase "ByteString" $ - let tests :: SymbioteT BS.ByteString IO () - tests = do - register "Generating ByteString" 50 (Proxy :: Proxy (Generating BS.ByteString)) - register "Operating ByteString" 50 (Proxy :: Proxy (Operating BS.ByteString)) - register "First ByteString" 50 (Proxy :: Proxy (First BS.ByteString)) - register "Second ByteString" 50 (Proxy :: Proxy (Second BS.ByteString)) - register "Topic" 50 (Proxy :: Proxy Topic) - in firstPeerZeroMQ "tcp://*:3002" NoDebug tests - ] + [ askOption $ \serverOrClient -> askOption $ \network -> case serverOrClient of + Server -> testGroup "WebSocket Server" + [ testCase "Json" $ + secondPeerWebSocketJson + WebSocketParams + { runWebSocket = \client -> do + let server = accept client + server' <- runServerAppT server + run 3000 $ logStdoutDev $ websocketsOr defaultConnectionOptions server' $ \_ respond -> + respond $ responseLBS status400 [] "Not a websocket" + , webSocketServerOrClient = WebSocketServer + , webSocketNetwork = network + } + NoDebug jsonTests + , testCase "ByteString" $ + secondPeerWebSocketByteString + WebSocketParams + { runWebSocket = \client -> do + let server = accept client + server' <- runServerAppT server + run 3001 $ logStdoutDev $ websocketsOr defaultConnectionOptions server' $ \_ respond -> + respond $ responseLBS status400 [] "Not a websocket" + , webSocketServerOrClient = WebSocketServer + , webSocketNetwork = network + } + NoDebug byteStringTests + ] + Client -> testGroup "WebSocket Client" + [ testCase "Json" $ + firstPeerWebSocketJson + WebSocketParams + { runWebSocket = \client -> do + runClient "localhost" 3000 "/" client + , webSocketServerOrClient = WebSocketClient + , webSocketNetwork = network + } + NoDebug jsonTests + , testCase "ByteString" $ + firstPeerWebSocketByteString + WebSocketParams + { runWebSocket = \client -> do + runClient "localhost" 3001 "/" client + , webSocketServerOrClient = WebSocketClient + , webSocketNetwork = network + } + NoDebug byteStringTests + ] + , askOption $ \serverOrClient -> askOption $ \network -> case serverOrClient of + Server -> testGroup "ZeroMQ Server" + [ testCase "ByteString" $ + secondPeerZeroMQ + ZeroMQParams + { zmqHost = "tcp://*:3002" + , zmqServerOrClient = ZeroMQServer + , zmqNetwork = network + } + NoDebug byteStringTests + ] + Client -> testGroup "ZeroMQ Client" + [ testCase "ByteString" $ + firstPeerZeroMQ + ZeroMQParams + { zmqHost = "tcp://127.0.0.1:3002" + , zmqServerOrClient = ZeroMQClient + , zmqNetwork = network + } + NoDebug byteStringTests + ] ] + where + jsonTests :: SymbioteT Json.Value IO () + jsonTests = do + register "Generating Topic" 100 (Proxy :: Proxy (Generating Topic)) + register "Operating Topic" 100 (Proxy :: Proxy (Operating Topic)) + register "First Topic" 100 (Proxy :: Proxy (First Topic)) + register "Second Topic" 100 (Proxy :: Proxy (Second Topic)) + register "Topic" 100 (Proxy :: Proxy Topic) + byteStringTests :: SymbioteT BS.ByteString IO () + byteStringTests = do + register "Generating ByteString" 50 (Proxy :: Proxy (Generating BS.ByteString)) + register "Operating ByteString" 50 (Proxy :: Proxy (Operating BS.ByteString)) + register "First ByteString" 50 (Proxy :: Proxy (First BS.ByteString)) + register "Second ByteString" 50 (Proxy :: Proxy (Second BS.ByteString)) + register "Topic" 50 (Proxy :: Proxy Topic) -- Internal instances instance SymbioteOperation (Generating s) (Generating s) where