|
| 1 | +[NotInParallel] |
| 2 | +public class AddFromIncomingExtraTests : |
| 3 | + IDisposable |
| 4 | +{ |
| 5 | + static readonly TestState state = new(); |
| 6 | + |
| 7 | + [Test] |
| 8 | + public async Task BufferSinkLetsTransformSeekTheOutput() |
| 9 | + { |
| 10 | + await Run<BufferSinkMessage>("BufferSink", "abcd"); |
| 11 | + await Assert.That(Encoding.UTF8.GetString(state.Bytes!)).IsEqualTo("abcd|len=4"); |
| 12 | + } |
| 13 | + |
| 14 | + [Test] |
| 15 | + public async Task MetadataIsAttachedToProducedOutgoingAttachment() |
| 16 | + { |
| 17 | + await Run<MetadataMessage>("Metadata", "ignored"); |
| 18 | + await Assert.That(state.Metadata).IsNotNull(); |
| 19 | + await Assert.That(state.Metadata!["k1"]).IsEqualTo("v1"); |
| 20 | + await Assert.That(state.Metadata["k2"]).IsEqualTo("v2"); |
| 21 | + } |
| 22 | + |
| 23 | + [Test] |
| 24 | + public async Task TransformExceptionSurfacesAndPreventsReply() |
| 25 | + { |
| 26 | + await Run<TransformThrowsMessage>("TransformThrows", "ignored", expectReply: false); |
| 27 | + await Assert.That(state.HandlerException).IsNotNull(); |
| 28 | + await Assert.That(state.HandlerException!.Message).Contains("transform boom"); |
| 29 | + } |
| 30 | + |
| 31 | + [Test] |
| 32 | + public async Task CleanupCallbackIsInvokedAfterCommit() |
| 33 | + { |
| 34 | + await Run<CleanupMessage>("Cleanup", "ignored"); |
| 35 | + await Assert.That(state.CleanupInvoked).IsTrue(); |
| 36 | + } |
| 37 | + |
| 38 | + public void Dispose() |
| 39 | + { |
| 40 | + } |
| 41 | + |
| 42 | + class TestState |
| 43 | + { |
| 44 | + public byte[]? Bytes; |
| 45 | + public IReadOnlyDictionary<string, string>? Metadata; |
| 46 | + public Exception? HandlerException; |
| 47 | + public bool CleanupInvoked; |
| 48 | + public ManualResetEvent Reply = new(false); |
| 49 | + } |
| 50 | + |
| 51 | + static async Task Run<TMessage>(string suffix, string sourceContent, bool expectReply = true) |
| 52 | + where TMessage : IMessage, new() |
| 53 | + { |
| 54 | + state.Bytes = null; |
| 55 | + state.Metadata = null; |
| 56 | + state.HandlerException = null; |
| 57 | + state.CleanupInvoked = false; |
| 58 | + state.Reply.Reset(); |
| 59 | + |
| 60 | + var attachmentsPath = Path.GetFullPath($"attachments/AddFromIncomingExtra_{suffix}"); |
| 61 | + if (Directory.Exists(attachmentsPath)) |
| 62 | + { |
| 63 | + Directory.Delete(attachmentsPath, recursive: true); |
| 64 | + } |
| 65 | + var transportPath = Path.GetFullPath($".learningtransport/AddFromIncomingExtra_{suffix}"); |
| 66 | + if (Directory.Exists(transportPath)) |
| 67 | + { |
| 68 | + Directory.Delete(transportPath, recursive: true); |
| 69 | + } |
| 70 | + |
| 71 | + var configuration = new EndpointConfiguration($"FileShareAddFromIncomingExtra_{suffix}"); |
| 72 | + configuration.UsePersistence<LearningPersistence>(); |
| 73 | + var transport = configuration.UseTransport<LearningTransport>(); |
| 74 | + transport.StorageDirectory(transportPath); |
| 75 | + configuration.RegisterComponents(_ => _.AddSingleton(state)); |
| 76 | + configuration.EnableAttachments(attachmentsPath, TimeToKeep.Default); |
| 77 | + configuration.UseSerialization<SystemJsonSerializer>(); |
| 78 | + configuration.DisableRetries(); |
| 79 | + var endpoint = await Endpoint.Start(configuration); |
| 80 | + |
| 81 | + var sendOptions = new SendOptions(); |
| 82 | + sendOptions.RouteToThisEndpoint(); |
| 83 | + var attachments = sendOptions.Attachments(); |
| 84 | + attachments.AddStream("input", async stream => |
| 85 | + { |
| 86 | + await using var writer = new StreamWriter(stream, leaveOpen: true); |
| 87 | + await writer.WriteAsync(sourceContent); |
| 88 | + }); |
| 89 | + await endpoint.Send(new TMessage(), sendOptions); |
| 90 | + |
| 91 | + var fired = state.Reply.WaitOne(TimeSpan.FromSeconds(expectReply ? 10 : 5)); |
| 92 | + await endpoint.Stop(); |
| 93 | + |
| 94 | + if (expectReply && !fired) |
| 95 | + { |
| 96 | + throw new("TimedOut waiting for reply"); |
| 97 | + } |
| 98 | + } |
| 99 | + |
| 100 | + class BufferSinkMessage : |
| 101 | + IMessage; |
| 102 | + |
| 103 | + class MetadataMessage : |
| 104 | + IMessage; |
| 105 | + |
| 106 | + class TransformThrowsMessage : |
| 107 | + IMessage; |
| 108 | + |
| 109 | + class CleanupMessage : |
| 110 | + IMessage; |
| 111 | + |
| 112 | + class OutMessage : |
| 113 | + IMessage; |
| 114 | + |
| 115 | + class BufferSinkHandler : |
| 116 | + IHandleMessages<BufferSinkMessage> |
| 117 | + { |
| 118 | + public Task Handle(BufferSinkMessage message, HandlerContext context) |
| 119 | + { |
| 120 | + var replyOptions = new ReplyOptions(); |
| 121 | + var outgoing = replyOptions.Attachments(); |
| 122 | + outgoing.AddFromIncoming( |
| 123 | + fromName: "input", |
| 124 | + toName: "output", |
| 125 | + bufferSink: true, |
| 126 | + transform: async (source, sink, cancel) => |
| 127 | + { |
| 128 | + if (!sink.CanSeek) |
| 129 | + { |
| 130 | + throw new("bufferSink: true expected the sink to be seekable."); |
| 131 | + } |
| 132 | + |
| 133 | + await source.CopyToAsync(sink, cancel); |
| 134 | + // Reading sink.Length is only valid when bufferSink: true (a MemoryStream). |
| 135 | + var trailer = Encoding.UTF8.GetBytes($"|len={sink.Length}"); |
| 136 | + await sink.WriteAsync(trailer, cancel); |
| 137 | + }); |
| 138 | + return context.Reply(new OutMessage(), replyOptions); |
| 139 | + } |
| 140 | + } |
| 141 | + |
| 142 | + class MetadataHandler : |
| 143 | + IHandleMessages<MetadataMessage> |
| 144 | + { |
| 145 | + public Task Handle(MetadataMessage message, HandlerContext context) |
| 146 | + { |
| 147 | + var replyOptions = new ReplyOptions(); |
| 148 | + var outgoing = replyOptions.Attachments(); |
| 149 | + outgoing.AddFromIncoming( |
| 150 | + fromName: "input", |
| 151 | + toName: "output", |
| 152 | + metadata: new Dictionary<string, string> |
| 153 | + { |
| 154 | + {"k1", "v1"}, |
| 155 | + {"k2", "v2"} |
| 156 | + }, |
| 157 | + transform: (source, sink, cancel) => source.CopyToAsync(sink, cancel)); |
| 158 | + return context.Reply(new OutMessage(), replyOptions); |
| 159 | + } |
| 160 | + } |
| 161 | + |
| 162 | + class TransformThrowsHandler : |
| 163 | + IHandleMessages<TransformThrowsMessage> |
| 164 | + { |
| 165 | + public async Task Handle(TransformThrowsMessage message, HandlerContext context) |
| 166 | + { |
| 167 | + try |
| 168 | + { |
| 169 | + var replyOptions = new ReplyOptions(); |
| 170 | + var outgoing = replyOptions.Attachments(); |
| 171 | + outgoing.AddFromIncoming( |
| 172 | + fromName: "input", |
| 173 | + toName: "output", |
| 174 | + transform: (_, _, _) => throw new InvalidOperationException("transform boom")); |
| 175 | + await context.Reply(new OutMessage(), replyOptions); |
| 176 | + } |
| 177 | + catch (Exception ex) |
| 178 | + { |
| 179 | + state.HandlerException = ex; |
| 180 | + state.Reply.Set(); |
| 181 | + } |
| 182 | + } |
| 183 | + } |
| 184 | + |
| 185 | + class CleanupHandler : |
| 186 | + IHandleMessages<CleanupMessage> |
| 187 | + { |
| 188 | + public Task Handle(CleanupMessage message, HandlerContext context) |
| 189 | + { |
| 190 | + var replyOptions = new ReplyOptions(); |
| 191 | + var outgoing = replyOptions.Attachments(); |
| 192 | + outgoing.AddFromIncoming( |
| 193 | + fromName: "input", |
| 194 | + toName: "output", |
| 195 | + cleanup: () => state.CleanupInvoked = true, |
| 196 | + transform: (source, sink, cancel) => source.CopyToAsync(sink, cancel)); |
| 197 | + return context.Reply(new OutMessage(), replyOptions); |
| 198 | + } |
| 199 | + } |
| 200 | + |
| 201 | + class OutHandler : |
| 202 | + IHandleMessages<OutMessage> |
| 203 | + { |
| 204 | + public async Task Handle(OutMessage message, HandlerContext context) |
| 205 | + { |
| 206 | + await using var memoryStream = new MemoryStream(); |
| 207 | + var incoming = context.Attachments(); |
| 208 | + await incoming.CopyTo("output", memoryStream, context.CancellationToken); |
| 209 | + state.Bytes = memoryStream.ToArray(); |
| 210 | + await foreach (var info in incoming.GetMetadata(context.CancellationToken)) |
| 211 | + { |
| 212 | + if (info.Name == "output") |
| 213 | + { |
| 214 | + state.Metadata = info.Metadata; |
| 215 | + } |
| 216 | + } |
| 217 | + state.Reply.Set(); |
| 218 | + } |
| 219 | + } |
| 220 | +} |
0 commit comments