eventstore 0.1.2.0 → 0.2.0.0

raw patch · 4 files changed

+113/−48 lines, 4 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• Database/EventStore.hs +3/−1
• Database/EventStore/Internal/Manager/Subscription.hs +102/−41
• Database/EventStore/Internal/Processor.hs +7/−5
• eventstore.cabal +1/−1

@@ -276,11 +276,13 @@ subscribe :: Connection           -> Text           -> Bool+          -> (Subscription -> Either DropReason ResolvedEvent -> IO ())           -> IO (Async Subscription)-subscribe Connection{..} stream_id res_lnk_tos = do+subscribe Connection{..} stream_id res_lnk_tos cb = do     tmp <- newEmptyTMVarIO     processorNewSubcription conProcessor                             (atomically . putTMVar tmp)+                            cb                             stream_id                             res_lnk_tos     async $ atomically $ readTMVar tmp

@@ -21,7 +21,9 @@     ) where  ---------------------------------------------------------------------------------import           Control.Concurrent.STM+import           Control.Concurrent+import           Control.Concurrent.Async+import           Control.Monad import           Control.Monad.Fix import           Data.ByteString (ByteString) import           Data.Int@@ -102,10 +104,17 @@ instance Decode SubscriptionDropped  --------------------------------------------------------------------------------+data UnsubscribeFromStream = UnsubscribeFromStream deriving (Generic, Show)++--------------------------------------------------------------------------------+instance Encode UnsubscribeFromStream++-------------------------------------------------------------------------------- data Pending     = Pending-      { penConfirmed :: Int64 -> Maybe Int32 -> IO Subscription-      , penCallback  :: Subscription -> IO ()+      { penConf  :: Int64 -> Maybe Int32 -> Subscription+      , penCb    :: Subscription -> IO ()+      , penEvtCb :: Subscription -> Either DropReason ResolvedEvent -> IO ()       }  --------------------------------------------------------------------------------@@ -121,17 +130,25 @@       { subId              :: !UUID       , subStream          :: !Text       , subResolveLinkTos  :: !Bool-      , subEventChan       :: !(TChan (Either DropReason ResolvedEventBuf))       , subLastCommitPos   :: !Int64       , subLastEventNumber :: !(Maybe Int32)       , subUnsubscribe     :: IO ()       }  --------------------------------------------------------------------------------+data Sub+    = Sub+      { _subSub  :: Subscription+      , _subCb   :: Subscription -> Either DropReason ResolvedEvent -> IO ()+      , _subChan :: Chan (IO ())+      , _subAS   :: Async ()+      }++-------------------------------------------------------------------------------- data Manager     = Manager       { _pendings      :: !(M.Map UUID Pending)-      , _subscriptions :: !(M.Map UUID Subscription)+      , _subscriptions :: !(M.Map UUID Sub)       }  --------------------------------------------------------------------------------@@ -172,7 +189,8 @@ data Appeared     = Appeared       { _appSub :: !Subscription-      , _appEvt :: !ResolvedEventBuf+      , _appEvt :: !ResolvedEvent+      , _appCb  :: Subscription -> ResolvedEvent -> IO ()       }  --------------------------------------------------------------------------------@@ -182,21 +200,36 @@           sub <- M.lookup packageCorrelation _subscriptions           sea <- maybeDecodeMessage packageData           let res_evt = getField $ streamResolvedEvent sea-              app     = Appeared sub res_evt+              chan    = _subChan sub+              evt_cb  = \s evt -> writeChan chan ((_subCb sub) s $ Right evt)+              app     = Appeared+                        { _appSub = _subSub sub+                        , _appEvt = newResolvedEventFromBuf res_evt+                        , _appCb  = evt_cb+                        }            return app     | otherwise = Nothing  ---------------------------------------------------------------------------------confirmSub :: Confirmation -> Manager -> IO (Maybe Subscription)+confirmSub :: Confirmation -> Manager -> IO (Maybe Sub) confirmSub (Confirmation uuid sc) Manager{..} =     for (M.lookup uuid _pendings) $ \p -> do         let last_com_pos = getField $ subscribeLastCommitPos sc             last_evt_num = getField $ subscribeLastEventNumber sc -        sub <- penConfirmed p last_com_pos last_evt_num-        penCallback p sub-        return sub+        let sub = penConf p last_com_pos last_evt_num+        penCb p sub+        chan <- newChan+        as   <- async $ forever $ do+            action <- readChan chan+            action+        return Sub+               { _subSub  = sub+               , _subCb   = penEvtCb p+               , _subChan = chan+               , _subAS   = as+               }  -------------------------------------------------------------------------------- -- Events@@ -207,13 +240,19 @@       , _subCallback       :: Subscription -> IO ()       , _subStream         :: !Text       , _subResolveLinkTos :: !Bool+      , _subEventAppeared  :: Subscription+                           -> Either DropReason ResolvedEvent+                           -> IO ()       }  -------------------------------------------------------------------------------- data Confirmation = Confirmation !UUID !SubscriptionConfirmation  ---------------------------------------------------------------------------------type NewSubscription = (Subscription -> IO ()) -> Text -> Bool -> IO ()+type NewSubscription =+    (Subscription -> IO ())+    -> (Subscription -> Either DropReason ResolvedEvent -> IO ())+    -> Text -> Bool -> IO ()  -------------------------------------------------------------------------------- subscriptionNetwork :: Settings@@ -229,33 +268,34 @@             on_con_sub = filterJust $ executeSyncIO $ snapshot confirmSub                                                                on_con                                                                mgr_b-            mgr_e = fmap (subscribe push_rem) on_sub <>-                    fmap remove on_rem               <>+            send_unsub = push_pkg . createUnsubscribePackage sett++            mgr_e = fmap (subscribe send_unsub) on_sub <>+                    fmap remove on_rem                 <>                     fmap confirmed on_con_sub          accum initManager mgr_e -    let on_app      = filterJust $ snapshot onEventAppeared e_pkg mgr_b-        on_drop     = filterJust $ execute $ snapshot (dropError push_rem)+    let on_app  = filterJust $ snapshot onEventAppeared e_pkg mgr_b+        on_drop = filterJust $ execute $ snapshot (dropError push_rem)                                                       e_pkg                                                       mgr_b         push_pkg_io = pushAsync push_pkg -        push_sub_io cb stream res_lnk_tos = randomIO >>= \uuid -> sync $+        push_sub_io cb evt_cb stream res_lnk_tos = randomIO >>= \uuid -> sync $             push_sub Subscribe                      { _subId             = uuid                      , _subCallback       = cb                      , _subStream         = stream                      , _subResolveLinkTos = res_lnk_tos+                     , _subEventAppeared  = evt_cb                      }      _ <- listen on_sub (push_pkg_io . createSubscribePackage sett) -    _ <- listen on_app $ \(Appeared sub evt) ->-        atomically $ writeTChan (subEventChan sub) (Right evt)+    _ <- listen on_app $ \(Appeared sub evt cb) -> cb sub evt -    _ <- listen on_drop $ \(Dropped reason sub _) ->-        atomically $ writeTChan (subEventChan sub) (Left reason)+    _ <- listen on_drop $ \(Dropped reason sub _ cb) -> cb sub reason      return push_sub_io @@ -272,11 +312,22 @@     msg = subscribeToStream _subStream _subResolveLinkTos  --------------------------------------------------------------------------------+createUnsubscribePackage :: Settings -> UUID -> Package+createUnsubscribePackage Settings{..} uuid =+    Package+    { packageCmd         = 0xC3+    , packageCorrelation = uuid+    , packageData        = runPut $ encodeMessage UnsubscribeFromStream+    , packageCred        = s_credentials+    }++-------------------------------------------------------------------------------- data Dropped     = Dropped       { droppedReason :: !DropReason       , droppedSub    :: !Subscription       , droppedId     :: !UUID+      , droppedCb     :: Subscription -> DropReason -> IO ()       }  --------------------------------------------------------------------------------@@ -293,10 +344,21 @@              sub <- M.lookup packageCorrelation _subscriptions              msg <- maybeDecodeMessage packageData              let reason  = fromMaybe D_Unsubscribed $ getField $ dropReason msg+                 chan    = _subChan sub++                 drop_cb = \s r -> do+                     var <- newEmptyMVar+                     writeChan chan $ do+                         (_subCb sub) s (Left r)+                         putMVar var ()+                     readMVar var+                     cancel $ _subAS sub+                  dropped = Dropped                            { droppedReason = reason-                           , droppedSub    = sub+                           , droppedSub    = _subSub sub                            , droppedId     = packageCorrelation+                           , droppedCb     = drop_cb                            }               return dropped@@ -311,30 +373,29 @@   where     pending =         Pending-        { penConfirmed = new_sub-        , penCallback  = _subCallback+        { penConf  = new_sub+        , penCb    = _subCallback+        , penEvtCb = _subEventAppeared         } -    new_sub com_pos last_evt = do-        chan <- newTChanIO-        let sub = Subscription-                  { subId              = _subId-                  , subStream          = _subStream-                  , subResolveLinkTos  = _subResolveLinkTos-                  , subEventChan       = chan-                  , subLastCommitPos   = com_pos-                  , subLastEventNumber = last_evt-                  , subUnsubscribe     = sync $ unsub _subId-                  }--        return sub+    new_sub com_pos last_evt =+        Subscription+        { subId              = _subId+        , subStream          = _subStream+        , subResolveLinkTos  = _subResolveLinkTos+        , subLastCommitPos   = com_pos+        , subLastEventNumber = last_evt+        , subUnsubscribe     = sync $ unsub _subId+        }  ---------------------------------------------------------------------------------confirmed :: Subscription -> Manager -> Manager-confirmed sub@Subscription{..} s@Manager{..} =-    s { _pendings      = M.delete subId _pendings-      , _subscriptions = M.insert subId sub _subscriptions+confirmed :: Sub -> Manager -> Manager+confirmed ss s@Manager{..} =+    s { _pendings      = M.delete sub_id _pendings+      , _subscriptions = M.insert sub_id ss _subscriptions       }+  where+    sub_id = subId $ _subSub ss  -------------------------------------------------------------------------------- remove :: UUID -> Manager -> Manager

@@ -55,10 +55,12 @@       { processorConnect        :: HostName -> Int -> IO ()       , processorShutdown       :: IO ()       , processorNewOperation   :: OperationParams -> IO ()-      , processorNewSubcription :: (Subscription -> IO ())-                                   -> Text-                                   -> Bool-                                   -> IO ()+      , processorNewSubcription ::+             (Subscription -> IO ())+             -> (Subscription -> Either DropReason ResolvedEvent -> IO ())+             -> Text+             -> Bool+             -> IO ()       }  --------------------------------------------------------------------------------@@ -176,7 +178,7 @@     let processor =             Processor             { processorConnect        = \h p -> sync $ pushConnect $ Connect h p-            , processorShutdown       = sync $ pushCleanup Cleanup+            , processorShutdown       = pushAsync pushCleanup Cleanup             , processorNewOperation   = \o -> sync $ push_new_op o             , processorNewSubcription = push_sub             }

@@ -10,7 +10,7 @@ -- PVP summary:      +-+------- breaking API changes --                   | | +----- non-breaking API additions --                   | | | +--- code changes with no API change-version:             0.1.2.0+version:             0.2.0.0  -- A short (one-line) description of the package. synopsis: EventStore Haskell TCP Client