finish message serialization of progress meters

Any given transfer can only display 1 progress meter at a time, or so
this code assumes. In some cases, there are progress meters for
different stages of a transfer, perhaps, and that is supported by this.

This commit was sponsored by Ethan Aubin.

Assistant/TransferrerPool.hs

@@ -11,6 +11,7 @@ import Assistant.Common
 import Assistant.Types.TransferrerPool
 import Types.Transfer
 import Utility.Batch
+import Messages.Serialized
 
 import qualified Command.TransferKeys as T
 
@@ -58,14 +59,9 @@ checkTransferrerPoolItem program batchmaker i = case i of
 performTransfer :: Transferrer -> Transfer -> TransferInfo -> Assistant Bool
 performTransfer transferrer t info = catchBoolIO $ do
 	(liftIO $ T.sendRequest t info (transferrerWrite transferrer))
-	readresponse
-  where
-	readresponse = 
-		liftIO (T.readResponse (transferrerRead transferrer)) >>= \case
-			Right r -> return r
-			Left so -> do
-				liftAnnex $ emitSerializedOutput so
-				readresponse
+	relaySerializedOutput
+		(liftIO (T.readResponse (transferrerRead transferrer)))
+		liftAnnex
 
 {- Starts a new git-annex transferkeys process, setting up handles
  - that will be used to communicate with it. -}

Messages.hs

@@ -50,7 +50,6 @@ module Messages (
 	withMessageState,
 	prompt,
 	mkPrompter,
-	emitSerializedOutput,
 ) where
 
 import System.Log.Logger

Messages/Internal.hs

@@ -96,18 +96,3 @@ flushed a = a >> hFlush stdout
 
 outputSerialized :: (SerializedOutput -> IO ()) -> SerializedOutput -> IO ()
 outputSerialized = id
-
-emitSerializedOutput :: SerializedOutput -> Annex ()
-emitSerializedOutput (OutputMessage msg) =
-	outputMessage' nojsonoutputter nojsonbuilder msg
-  where
-	nojsonoutputter _ _ = return False
-	nojsonbuilder = id
-emitSerializedOutput (OutputError msg) = outputError msg
-emitSerializedOutput (ProgressMeter sz old new) = undefined -- TODO
-emitSerializedOutput (JSONObject b) =
-	withMessageState $ \s -> case outputType s of
-		JSONOutput _ -> liftIO $ flushed $ JSON.emit' b
-		SerializedOutput h -> liftIO $
-			outputSerialized h $ JSONObject b
-		_ -> q

Messages/Progress.hs

@@ -84,10 +84,10 @@ metered'
 	-- ^ this should run showOutput
 	-> (Meter -> MeterUpdate -> m a)
 	-> m a
-metered' st othermeter size showoutput a = go size st
+metered' st othermeter msize showoutput a = go st
   where
-	go _ (MessageState { outputType = QuietOutput }) = nometer
-	go msize (MessageState { outputType = NormalOutput, concurrentOutputEnabled = False }) = do
+	go (MessageState { outputType = QuietOutput }) = nometer
+	go (MessageState { outputType = NormalOutput, concurrentOutputEnabled = False }) = do
 		showoutput
 		meter <- liftIO $ mkMeter msize $ 
 			displayMeterHandle stdout bandwidthMeter
@@ -96,7 +96,7 @@ metered' st othermeter size showoutput a = go size st
 		r <- a meter (combinemeter m)
 		liftIO $ clearMeterHandle meter stdout
 		return r
-	go msize (MessageState { outputType = NormalOutput, concurrentOutputEnabled = True }) =
+	go (MessageState { outputType = NormalOutput, concurrentOutputEnabled = True }) =
 		withProgressRegion st $ \r -> do
 			meter <- liftIO $ mkMeter msize $ \_ msize' old new ->
 				let s = bandwidthMeter msize' old new
@@ -104,7 +104,7 @@ metered' st othermeter size showoutput a = go size st
 			m <- liftIO $ rateLimitMeterUpdate consoleratelimit meter $
 				updateMeter meter
 			a meter (combinemeter m)
-	go msize (MessageState { outputType = JSONOutput jsonoptions })
+	go (MessageState { outputType = JSONOutput jsonoptions })
 		| jsonProgress jsonoptions = do
 			let buf = jsonBuffer st
 			meter <- liftIO $ mkMeter msize $ \_ msize' _old new ->
@@ -113,12 +113,15 @@ metered' st othermeter size showoutput a = go size st
 				updateMeter meter
 			a meter (combinemeter m)
 		| otherwise = nometer
-	go msize (MessageState { outputType = SerializedOutput h }) = do
-		meter <- liftIO $ mkMeter msize $ \_ msize' old new ->
-			outputSerialized h $ ProgressMeter msize' old new
+	go (MessageState { outputType = SerializedOutput h }) = do
+		liftIO $ outputSerialized h $ StartProgressMeter msize
+		meter <- liftIO $ mkMeter msize $ \_ _ _old new ->
+			outputSerialized h $ UpdateProgressMeter $
+				meterBytesProcessed new
 		m <- liftIO $ rateLimitMeterUpdate minratelimit meter $
 			updateMeter meter
 		a meter (combinemeter m)
+			`finally` (liftIO $ outputSerialized h EndProgressMeter)
 	nometer = do
 		dummymeter <- liftIO $ mkMeter Nothing $
 			\_ _ _ _ -> return ()

Messages/Serialized.hs

@@ -0,0 +1,71 @@
+{- serialized output
+ -
+ - Copyright 2020 Joey Hess <id@joeyh.name>
+ -
+ - Licensed under the GNU AGPL version 3 or higher.
+ -}
+
+{-# LANGUAGE RankNTypes #-}
+
+module Messages.Serialized (outputSerialized, relaySerializedOutput) where
+
+import Common
+import Annex
+import Types.Messages
+import Messages
+import Messages.Internal
+import Messages.Progress
+import qualified Messages.JSON as JSON
+
+import Control.Monad.IO.Class (MonadIO)
+
+relaySerializedOutput
+	:: (Monad m, MonadIO m, MonadMask m)
+	=> m (Either SerializedOutput r)
+	-- ^ Get next serialized output, or final value to return.
+	-> (forall a. Annex a -> m a)
+	-- ^ Run an annex action in the monad. Will not be used with
+	-- actions that block for a long time.
+	-> m r
+relaySerializedOutput getso runannex = go Nothing
+  where
+	go st = loop st >>= \case
+		Right r -> return r
+		Left st' -> go st'
+	
+	loop st = getso >>= \case
+		Right r -> return (Right r)
+		Left (OutputMessage msg) -> do
+			runannex $ outputMessage'
+				(\_ _ -> return False)
+				id
+				msg
+			loop st
+		Left (OutputError msg) -> do
+			runannex $ outputError msg
+			loop st		
+		Left (JSONObject b) -> do
+			runannex $ withMessageState $ \s -> case outputType s of
+				JSONOutput _ -> liftIO $ flushed $ JSON.emit' b
+				SerializedOutput h -> liftIO $
+					outputSerialized h $ JSONObject b
+				_ -> q
+			loop st
+		Left (StartProgressMeter sz) -> do
+			ost <- runannex (Annex.getState Annex.output)
+			-- Display a progress meter while running, until
+			-- the meter ends or a final value is returned.
+			metered' ost Nothing sz (runannex showOutput) 
+				(\_meter meterupdate -> loop (Just meterupdate))
+				>>= \case
+					Right r -> return (Right r)
+					-- Continue processing serialized
+					-- output after the progress meter
+					-- is done.
+					Left _st' -> loop Nothing
+		Left EndProgressMeter -> return (Left st)
+		Left (UpdateProgressMeter n) -> do
+			case st of
+				Just meterupdate -> liftIO $ meterupdate n
+				Nothing -> noop
+			loop st

Types/Messages.hs

@@ -9,6 +9,7 @@ module Types.Messages where
 
 import qualified Utility.Aeson as Aeson
 import Utility.Metered
+import Utility.FileSize
 
 import Control.Concurrent
 import System.Console.Regions (ConsoleRegion)
@@ -66,7 +67,9 @@ newMessageState = do
 data SerializedOutput
 	= OutputMessage S.ByteString
 	| OutputError String
-	| ProgressMeter (Maybe Integer) MeterState MeterState
+	| StartProgressMeter (Maybe FileSize)
+	| UpdateProgressMeter BytesProcessed
+	| EndProgressMeter
 	| JSONObject L.ByteString
 	-- ^ This is always sent, it's up to the consumer to decide if it
 	-- wants to display JSON, or human-readable messages.

git-annex.cabal

@@ -931,6 +931,7 @@ Executable git-annex
     Messages.Internal
     Messages.JSON
     Messages.Progress
+    Messages.Serialized
     P2P.Address
     P2P.Annex
     P2P.Auth