packages feed

symbiote 0.0.2 → 0.0.3

raw patch · 12 files changed

+1076/−375 lines, 12 filesdep +chandep +hashabledep +threaded

Dependencies added: chan, hashable, threaded, uuid

Files

LICENSE view
@@ -1,4 +1,4 @@-Copyright Athen Clark (c) 2019+Copyright Athen Clark (c) 2019, 2020  All rights reserved.
src/Test/Serialization/Symbiote.hs view
@@ -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)
src/Test/Serialization/Symbiote/Core.hs view
@@ -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 <https://symbiotic-data.github.io/#/data/?id=string32 symbiotic-data standard>. 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 <http://hackage.haskell.org/package/QuickCheck-2.13.2/docs/Test-QuickCheck-Gen.html#v:sized size> 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 <http://hackage.haskell.org/package/QuickCheck-2.13.2/docs/Test-QuickCheck-Gen.html#v:sized size> 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)
src/Test/Serialization/Symbiote/Debug.hs view
@@ -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)
src/Test/Serialization/Symbiote/WebSocket.hs view
@@ -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+  }
+ src/Test/Serialization/Symbiote/WebSocket/Ident.hs view
@@ -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
src/Test/Serialization/Symbiote/ZeroMQ.hs view
@@ -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)
symbiote.cabal view
@@ -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 <https://github.com/athanclark/symbiote#readme> 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
+ test/ServerOrClient.hs view
@@ -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"
test/Spec.hs view
@@ -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" $
test/Spec/Local.hs view
@@ -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))             ]           ]     ]
test/Spec/Protocol.hs view
@@ -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