dont always get the bytes

Author: LukeButters

source/Halibut.Tests/Transport/Protocol/MessageSerializerFixture.cs

@@ -305,7 +305,7 @@ static byte[] DeflateString(string s)
 
         async Task<T> ReadMessage<T>(MessageSerializer messageSerializer, RewindableBufferStream rewindableBufferStream)
         {
-            return (await messageSerializer.ReadMessageAsync<T>(rewindableBufferStream, CancellationToken)).Message;
+            return (await messageSerializer.ReadMessageAsync<T>(rewindableBufferStream, false, CancellationToken)).Message;
         }
 
         async Task WriteMessage(MessageSerializer messageSerializer, Stream stream, string message)

source/Halibut.Tests/Transport/Protocol/ProtocolFixture.cs

@@ -388,7 +388,7 @@ public async Task SendAsync<T>(T message, CancellationToken cancellationToken)
                 output.AppendLine("--> " + typeof(T).Name);
             }
 
-            public Task SendAsync(PreparedRequestMessage preparedRequestMessage, CancellationToken cancellationToken)
+            public Task SendPrePreparedRequestAsync(PreparedRequestMessage preparedRequestMessage, CancellationToken cancellationToken)
             {
                 throw new NotImplementedException();
             }

source/Halibut/Transport/Protocol/IMessageExchangeStream.cs

@@ -36,7 +36,7 @@ public interface IMessageExchangeStream
 
         Task SendAsync<T>(T message, CancellationToken cancellationToken);
 
-        Task SendAsync(PreparedRequestMessage preparedRequestMessage, CancellationToken cancellationToken);
+        Task SendPrePreparedRequestAsync(PreparedRequestMessage preparedRequestMessage, CancellationToken cancellationToken);
 
         Task<RequestMessage?> ReceiveRequestAsync(TimeSpan timeoutForReceivingTheFirstByte, CancellationToken cancellationToken);
         Task<ResponseMessage?> ReceiveResponseAsync(CancellationToken cancellationToken);

source/Halibut/Transport/Protocol/IMessageSerializer.cs

@@ -10,6 +10,7 @@ namespace Halibut.Transport.Protocol
     public interface IMessageSerializer
     {
         Task<IReadOnlyList<DataStream>> WriteMessageAsync<T>(Stream stream, T message, CancellationToken cancellationToken);
-        Task<(T Message, IReadOnlyList<DataStream> DataStreams, byte[]? CompressedMessageBytes)> ReadMessageAsync<T>(RewindableBufferStream stream, CancellationToken cancellationToken);
+        Task<(T Message, IReadOnlyList<DataStream> DataStreams, byte[]? CompressedMessageBytes)> ReadMessageAsync<T>(
+            RewindableBufferStream stream, bool captureData, CancellationToken cancellationToken);
     }
 }

source/Halibut/Transport/Protocol/MessageExchangeProtocol.cs

@@ -307,7 +307,7 @@ async Task<ResponseMessage> SendAndReceiveRequest(RequestMessage nextRequest, Ca
 
         async Task<ResponseBytesAndDataStreams> SendAndReceiveRequest(PreparedRequestMessage preparedRequestMessage, CancellationToken cancellationToken)
         {
-            await stream.SendAsync(preparedRequestMessage, cancellationToken);
+            await stream.SendPrePreparedRequestAsync(preparedRequestMessage, cancellationToken);
             return (await stream.ReceiveResponseBytesAsync(cancellationToken))!;
         }
 

source/Halibut/Transport/Protocol/MessageExchangeStream.cs

@@ -195,7 +195,7 @@ public async Task SendAsync<T>(T message, CancellationToken cancellationToken)
             log.Write(EventType.Diagnostic, "Sent: {0}", message);
         }
 
-        public async Task SendAsync(PreparedRequestMessage preparedRequestMessage, CancellationToken cancellationToken)
+        public async Task SendPrePreparedRequestAsync(PreparedRequestMessage preparedRequestMessage, CancellationToken cancellationToken)
         {
             await stream.WriteAsync(preparedRequestMessage.RequestBytes, cancellationToken);
             await WriteEachStreamAsync(preparedRequestMessage.DataStreams, cancellationToken);
@@ -224,7 +224,7 @@ await stream.WithReadTimeout(
 
         public async Task<ResponseBytesAndDataStreams?> ReceiveResponseBytesAsync(CancellationToken cancellationToken)
         {
-            var (result, dataStreams, compressedMessageBytes) = await serializer.ReadMessageAsync<ResponseMessage>(stream, cancellationToken);
+            var (result, dataStreams, compressedMessageBytes) = await serializer.ReadMessageAsync<ResponseMessage>(stream, true, cancellationToken);
             await ReadStreamsAsync(dataStreams, cancellationToken);
             log.Write(EventType.Diagnostic, "Received: {0}", result); // TODO stop sending the response to logs.
             if (compressedMessageBytes == null) return null;
@@ -233,7 +233,7 @@ await stream.WithReadTimeout(
 
         async Task<T?> ReceiveAsync<T>(CancellationToken cancellationToken)
         {
-            var (result, dataStreams, compressedMessageBytes) = await serializer.ReadMessageAsync<T>(stream, cancellationToken);
+            var (result, dataStreams, compressedMessageBytes) = await serializer.ReadMessageAsync<T>(stream, false, cancellationToken);
             await ReadStreamsAsync(dataStreams, cancellationToken);
             log.Write(EventType.Diagnostic, "Received: {0}", result); // TODO stop sending the response to logs.
             return result;

source/Halibut/Transport/Protocol/MessageSerializer.cs

@@ -64,15 +64,17 @@ public async Task<IReadOnlyList<DataStream>> WriteMessageAsync<T>(Stream stream,
             return serializedStreams;
         }
 
-        public async Task<(T Message, IReadOnlyList<DataStream> DataStreams, byte[]? CompressedMessageBytes)> ReadMessageAsync<T>(RewindableBufferStream stream, CancellationToken cancellationToken)
+        public async Task<(T Message, IReadOnlyList<DataStream> DataStreams, byte[]? CompressedMessageBytes)> ReadMessageAsync<T>(
+            RewindableBufferStream stream, 
+            bool captureData,
+            CancellationToken cancellationToken)
         {
             await using (var errorRecordingStream = new ErrorRecordingStream(stream, closeInner: false))
             {
                 Exception? exceptionFromDeserialisation = null;
                 try
                 {
-                    // TODO don't always set true.
-                    return await ReadCompressedMessageAsync<T>(errorRecordingStream, stream, true, cancellationToken);
+                    return await ReadCompressedMessageAsync<T>(errorRecordingStream, stream, captureData, cancellationToken);
                 }
                 catch (Exception e)
                 {