Deferred object loading

2 files changed, 51 insertions, 9 deletions

diff --git a/main/Test.hs b/main/Test.hs
index 5b6509a..6896c9a 100644
--- a/main/Test.hs
+++ b/main/Test.hs
@@ -42,6 +42,7 @@ import Erebos.Identity
 import Erebos.Invite
 import Erebos.Network
 import Erebos.Object
+import Erebos.Object.Deferred
 import Erebos.Pairing
 import Erebos.PubKey
 import Erebos.Service
@@ -70,6 +71,7 @@ data RunningServer = RunningServer
     { rsServer :: Server
     , rsPeers :: MVar ( Int, [ TestPeer ] )
     , rsPeerThread :: ThreadId
+    , rsDeferredObjects :: MVar [ Deferred Object ]
     }
 
 data TestPeer = TestPeer
@@ -284,6 +286,7 @@ commands =
     , ( "store-raw", cmdStoreRaw )
     , ( "load", cmdLoad )
     , ( "load-type", cmdLoadType )
+    , ( "load-deferred", cmdLoadDeferred )
     , ( "stored-generation", cmdStoredGeneration )
     , ( "stored-roots", cmdStoredRoots )
     , ( "stored-set-add", cmdStoredSetAdd )
@@ -392,15 +395,36 @@ cmdLoadType :: Command
 cmdLoadType = do
     st <- asks tiStorage
     [ tref ] <- asks tiParams
-    Just ref <- liftIO $ readRef st $ encodeUtf8 tref
-    let obj = load @Object ref
-    let otype = case obj of
-            Blob {} -> "blob"
-            Rec {} -> "rec"
-            OnDemand {} -> "ondemand"
-            ZeroObject {} -> "zero"
-            UnknownObject utype _ -> "unknown " <> decodeUtf8 utype
-    cmdOut $ "load-type " <> T.unpack otype
+    liftIO (readRef st $ encodeUtf8 tref) >>= \case
+        Just ref -> do
+            let obj = load @Object ref
+            let otype = case obj of
+                    Blob {} -> "blob"
+                    Rec {} -> "rec"
+                    OnDemand {} -> "ondemand"
+                    ZeroObject {} -> "zero"
+                    UnknownObject utype _ -> "unknown " <> decodeUtf8 utype
+            cmdOut $ "load-type " <> T.unpack otype
+        Nothing -> do
+            cmdOut $ "load-type-failed"
+
+cmdLoadDeferred :: Command
+cmdLoadDeferred = do
+    st <- asks tiStorage
+    [ tidx ] <- asks tiParams
+    Just RunningServer {..} <- gets tsServer
+    deferred <- (!! read (T.unpack tidx)) <$> liftIO (readMVar rsDeferredObjects)
+    mvar <- deferredLoad deferred
+    out <- asks tiOutput
+    liftIO $ void $ forkIO $ readMVar mvar >>= \case
+        DeferredLoaded sobj -> do
+            void $ copyRef st $ storedRef sobj
+            header : _ <- return $ BL.lines $ serializeObject $ fromStored sobj
+            outLine out $ T.unpack $ T.unwords [ "load-deferred-done", tidx, decodeUtf8 $ BL.toStrict header ]
+        DeferredInvalid -> do
+            outLine out $ T.unpack $ T.unwords [ "load-deferred-invalid", tidx ]
+        DeferredFailed -> do
+            outLine out $ T.unpack $ T.unwords [ "load-deferred-failed", tidx ]
 
 cmdStoredGeneration :: Command
 cmdStoredGeneration = do
@@ -581,6 +605,7 @@ cmdStartServer = do
 
     h <- getOrLoadHead
     rsPeers <- liftIO $ newMVar (1, [])
+    rsDeferredObjects <- liftIO $ newMVar []
     services <- forM serviceNames $ \case
         ( "attach", _ ) -> return $ someServiceAttr $ pairingAttributes (Proxy @AttachService) out rsPeers "attach"
         ( "chatroom", _ ) -> return $ someService @ChatroomService Proxy
@@ -609,6 +634,10 @@ cmdStartServer = do
                                 StreamClosed seqNum -> do
                                     outLine out $ unwords [ "test-stream-closed-from", show pidx, show num, show seqNum ]
                         go
+            , testOnDemandReceived = \size deferred -> do
+                liftIO $ do
+                    idx <- modifyMVar rsDeferredObjects (\ds -> return ( ds ++ [ deferred ], length ds ))
+                    outLine out $ unwords [ "test-ondemand-received", show idx, show size, show $ deferredRef deferred ]
             }
         ( sname, _ ) -> throwOtherError $ "unknown service `" <> T.unpack sname <> "'"
 
diff --git a/main/Test/Service.hs b/main/Test/Service.hs
index c0be07d..156b62c 100644
--- a/main/Test/Service.hs
+++ b/main/Test/Service.hs
@@ -9,9 +9,12 @@ import Control.Monad
 import Control.Monad.Reader
 
 import Data.ByteString.Lazy.Char8 qualified as BL
+import Data.Word
 
+import Erebos.Identity
 import Erebos.Network
 import Erebos.Object
+import Erebos.Object.Deferred
 import Erebos.Service
 import Erebos.Service.Stream
 import Erebos.Storable
@@ -21,6 +24,7 @@ data TestMessage = TestMessage (Stored Object)
 data TestMessageAttributes = TestMessageAttributes
     { testMessageReceived :: Object -> String -> String -> String -> ServiceHandler TestMessage ()
     , testStreamsReceived :: [ StreamReader ] -> ServiceHandler TestMessage ()
+    , testOnDemandReceived :: Word64 -> Deferred Object -> ServiceHandler TestMessage ()
     }
 
 instance Storable TestMessage where
@@ -34,6 +38,7 @@ instance Service TestMessage where
     defaultServiceAttributes _ = TestMessageAttributes
         { testMessageReceived = \_ _ _ _ -> return ()
         , testStreamsReceived = \_ -> return ()
+        , testOnDemandReceived = \_ _ -> return ()
         }
 
     serviceHandler smsg = do
@@ -50,6 +55,14 @@ instance Service TestMessage where
             cb <- asks $ testStreamsReceived . svcAttributes
             cb streams
 
+        case obj of
+            OnDemand size dgst -> do
+                cb <- asks $ testOnDemandReceived . svcAttributes
+                server <- asks svcServer
+                pid <- asks svcPeerIdentity
+                cb size =<< liftIO (deferLoadWithServer dgst server [ refDigest $ storedRef $ idData pid ])
+            _ -> return ()
+
 
 openTestStreams :: Int -> ServiceHandler TestMessage [ StreamWriter ]
 openTestStreams count = do