support configurable output data writers in recorder

Runtime/Core/Recorder/Module/Frame/FrameRecorderModule.cs

@@ -61,6 +61,7 @@ void IRecorderModule.Awake(RecorderContext ctx)
         void IRecorderModule.StartRecording(RecorderContext ctx)
         {
             _lastUpdateTime = 0;
+            _lastFixedUpdateTime = ctx.CurrentRecord.Time;
             _shouldRunUpdate = true;
             _shouldSerialize = true;
 

Runtime/Core/Recorder/Recorder.cs

@@ -15,6 +15,7 @@
 using ProtoBurst;
 using Unity.Collections;
 using UnityEngine;
+using UnityEngine.Assertions;
 using UnityEngine.Pool;
 using static PLUME.Core.Utils.SampleUtils;
 
@@ -49,7 +50,7 @@ private PlumeRecorder(DataDispatcher dataDispatcher, RecorderContext ctx)
         /// Starts the recording process. If the recorder is already recording, throw a <see cref="InvalidOperationException"/> exception.
         /// </summary>
         /// <exception cref="InvalidOperationException"></exception>
-        private void StartRecordingInternal(string name, string extraMetadata = "")
+        private void StartRecordingInternal(string name, string extraMetadata = "", IDataWriter[] outputs = null)
         {
             if (_context.Status is RecorderStatus.Stopping)
                 throw new InvalidOperationException(
@@ -74,7 +75,24 @@ private void StartRecordingInternal(string name, string extraMetadata = "")
             RecordApplicationGlobalSettings(record);
             RecordApplicationCurrentSettings(record);
 
-            _dataDispatcher.Start(_context.CurrentRecord);
+            if (outputs == null)
+            {
+                var outputDir = Application.persistentDataPath;
+
+                FileDataWriter.GenerateFilePath(outputDir, name, out string filePath, out string metaFilePath);
+
+                IDataWriter fileDataWriter = new FileDataWriter(filePath, metaFilePath);
+                outputs = new IDataWriter[] { fileDataWriter };
+            }
+
+            Assert.IsFalse(outputs.Length == 0, "The outputs length is 0");
+
+            foreach (IDataWriter output in outputs)
+            {
+                output.SetMetaData(recordMetadata);
+            }
+
+            _dataDispatcher.Start(_context.CurrentRecord, outputs);
 
             // ReSharper disable once ForCanBeConvertedToForeach
             for (var i = 0; i < _context.Modules.Count; i++)
@@ -272,6 +290,7 @@ private void Awake()
 
             if (recorderSettings.StartOnPlay)
             {
+                // TODO: allow configuring data writers in Settings
                 StartRecordingInternal(recorderSettings.DefaultRecordPrefix, recorderSettings.DefaultRecordExtraMetadata);
             }
         }
@@ -293,4 +312,4 @@ public void Dispose()
                 module.Destroy(_context);
         }
     }
-}
+}

Runtime/Core/Recorder/RecorderExtensions.cs

@@ -29,10 +29,10 @@ private static void CheckInstantiated()
                 throw new InvalidOperationException("PLUME recorder instance is not created yet.");
         }
 
-        public static void StartRecording(string name, string extraMetadata = "")
+        public static void StartRecording(string name, string extraMetadata = "", IDataWriter[] writers = null)
         {
             CheckInstantiated();
-            Instance.StartRecordingInternal(name, extraMetadata);
+            Instance.StartRecordingInternal(name, extraMetadata, writers);
         }
 
         public static async UniTask StopRecording()
@@ -127,4 +127,4 @@ internal static void OnSceneLoaded()
             Instance.Awake();
         }
     }
-}
+}

Runtime/Core/Recorder/Writer/DataDispatcher.cs

@@ -14,13 +14,9 @@ public class DataDispatcher
         private IDataWriter[] _outputs;
         private NetworkStream _networkStream;
 
-        internal void Start(Record record)
+        internal void Start(Record record, IDataWriter[] outputs)
         {
-            var fileDataWriter = new FileDataWriter(record);
-            _outputs = new IDataWriter[] { fileDataWriter };
-
-            // var networkDataWriter = new NetworkDataWriter(recordIdentifier);
-            // _outputs = new IDataWriter[] { networkDataWriter };
+            _outputs = outputs;
 
             _dispatcherThread = new Thread(() => DispatchLoop(record))
             {
@@ -150,4 +146,4 @@ public void OnApplicationPaused()
             }
         }
     }
-}
+}

Runtime/Core/Recorder/Writer/FileDataWriter.cs

@@ -7,6 +7,7 @@
 using K4os.Compression.LZ4.Streams;
 using PLUME.Sample;
 using UnityEngine;
+using UnityEngine.Assertions;
 
 namespace PLUME.Core.Recorder.Writer
 {
@@ -19,15 +20,16 @@ public class FileDataWriter : IDataWriter, IDisposable
         private readonly Stream _metaStream;
         private readonly CodedOutputStream _metaCodedOutputStream;
 
-        private readonly Sample.RecordMetadata _metadata;
         private readonly RecordMetrics _metrics;
-
-        public FileDataWriter(Record record)
+        private Sample.RecordMetadata _metadata;
+
+        /// <param name="filePath">The file path for the plm file.</param>
+        /// <param name="metaFilePath">The file path for the meta file.</param>
+        public FileDataWriter(string filePath, string metaFilePath)
         {
-            var outputDir = Application.persistentDataPath;
+            Assert.IsFalse(string.IsNullOrWhiteSpace(filePath), "filePath is null or empty");
+            Assert.IsFalse(string.IsNullOrWhiteSpace(metaFilePath), "metaFilePath is null or empty");
 
-            GenerateFilePath(outputDir, record.Metadata, out var filePath, out var metaFilePath);
-
             Logger.Log($"Record will be saved to '{filePath}'.");
 
             PinnedMemory.MaxPooledSize = 0;
@@ -48,22 +50,25 @@ public FileDataWriter(Record record)
             _metaStream = File.Create(metaFilePath);
             _metaCodedOutputStream = new CodedOutputStream(_metaStream);
 
-            _metadata = record.Metadata.ToPayload();
             _metrics = new RecordMetrics
             {
                 IsSequential = true
             };
-
+        }
+
+        public void SetMetaData(RecordMetadata metaData)
+        {
+            _metadata = metaData.ToPayload();
             UpdateMetaFile();
         }
 
-        private static void GenerateFilePath(string outputDir, RecordMetadata recordMetadata,
+        public static void GenerateFilePath(string outputDir, string recordMetadataName,
             out string filePath, out string metadataPath)
         {
             var invalidChars = Path.GetInvalidFileNameChars().ToList();
             invalidChars.Add(' ');
 
-            var name = recordMetadata.Name;
+            var name = recordMetadataName;
             var safeName = new string(name.Select(c => invalidChars.Contains(c) ? '_' : c).ToArray());
 
             var formattedDateTime = DateTime.UtcNow.ToString("yyyy-MM-ddTHH-mm-sszz");
@@ -103,6 +108,10 @@ public void WriteTimestampedData(DataChunksTimestamped dataChunks)
 
         private void UpdateMetaFile()
         {
+            if (_metadata == null)
+            {
+                return;
+            }
             _metaStream.SetLength(0);
             _metaStream.Position = 0;
             _metaCodedOutputStream.WriteLength(_metadata.CalculateSize());
@@ -133,4 +142,4 @@ public void Dispose()
             _metaCodedOutputStream.Dispose();
         }
     }
-}
+}

Runtime/Core/Recorder/Writer/IDataWriter.cs

@@ -2,6 +2,8 @@ namespace PLUME.Core.Recorder.Writer
 {
     public interface IDataWriter
     {
+        public void SetMetaData(RecordMetadata metadata);
+
         public void WriteTimelessData(DataChunks dataChunks);
 
         public void WriteTimestampedData(DataChunksTimestamped dataChunks);
@@ -10,4 +12,4 @@ public interface IDataWriter
 
         public void Close();
     }
-}
+}

Runtime/Core/Recorder/Writer/NetworkDataWriter.cs

@@ -12,10 +12,10 @@ public class NetworkDataWriter : IDataWriter
     {
         private readonly Stream _stream;
 
-        public NetworkDataWriter(RecordMetadata recordMetadata)
+        public NetworkDataWriter(string ipAddress="127.0.0.1", int port=8000)
         {
             // Create a tcp server
-            var server = new TcpListener(IPAddress.Parse("127.0.0.1"), 8000);
+            var server = new TcpListener(IPAddress.Parse(ipAddress), port);
             server.Start();
 
             var stream = server.AcceptTcpClient().GetStream();
@@ -29,6 +29,9 @@ public NetworkDataWriter(RecordMetadata recordMetadata)
             _stream = LZ4Stream.Encode(stream, LZ4Level.L00_FAST);
         }
 
+        public void SetMetaData(RecordMetadata _)
+        {}
+
         public void WriteTimelessData(DataChunks dataChunks)
         {
         }
@@ -50,4 +53,4 @@ public void Close()
             _stream.Close();
         }
     }
-}
+}