Skip to content

Commit fc15b4f

Browse files
nblumhardtnblumhardt
authored
Merge pull request #11 from datalust/update-to-rmq7
Update to RabbitMQ.Client 7, lift other dependency versions
2 parents 6619f9f + e6843e4 commit fc15b4f

File tree

7 files changed

+206
-208
lines changed

7 files changed

+206
-208
lines changed

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# Seq.Input.RabbitMQ [![CI](https://github.com/datalust/seq-input-rabbitmq/actions/workflows/ci.yml/badge.svg)](https://github.com/datalust/seq-input-rabbitmq/actions/workflows/ci.yml)
22

3-
A Seq custom input that pulls events from RabbitMQ. **Requires Seq 5.1+.**
3+
A Seq custom input that pulls events from RabbitMQ. **Requires Seq 2025.2+.**
44

55
### Getting started
66

‎example/Demo/Demo.csproj‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,9 @@
66
</PropertyGroup>
77

88
<ItemGroup>
9-
<PackageReference Include="Serilog" Version="2.8.0" />
10-
<PackageReference Include="Serilog.Sinks.RabbitMQ" Version="2.0.2" />
11-
<PackageReference Include="Serilog.Formatting.Compact" Version="1.0.0" />
9+
<PackageReference Include="Serilog" Version="4.3.0" />
10+
<PackageReference Include="Serilog.Sinks.RabbitMQ" Version="8.0.0" />
11+
<PackageReference Include="Serilog.Formatting.Compact" Version="3.0.0" />
1212
</ItemGroup>
1313

1414
</Project>

‎example/Demo/Program.cs‎

Lines changed: 16 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,34 +1,22 @@
1-
using System;
2-
using System.Threading;
1+
using System.Threading;
32
using Serilog;
43
using Serilog.Formatting.Compact;
5-
using Serilog.Sinks.RabbitMQ.Sinks.RabbitMQ;
64

7-
namespace Demo
8-
{
9-
public class Program
5+
Log.Logger = new LoggerConfiguration()
6+
.Enrich.WithProperty("Application", "Demo")
7+
.WriteTo.RabbitMQ((client, sink) =>
108
{
11-
public static void Main()
12-
{
13-
var rmq = new RabbitMQConfiguration
14-
{
15-
Hostname = "localhost",
16-
Username = "guest",
17-
Password = "guest",
18-
Exchange = "",
19-
RouteKey = "logs"
20-
};
21-
22-
Log.Logger = new LoggerConfiguration()
23-
.Enrich.WithProperty("Application", "Demo")
24-
.WriteTo.RabbitMQ(rmq, new CompactJsonFormatter())
25-
.CreateLogger();
9+
client.Hostnames.Add("localhost");
10+
client.Username = "guest";
11+
client.Password = "guest";
12+
client.Exchange = "";
13+
client.RoutingKey = "logs";
14+
sink.TextFormatter = new CompactJsonFormatter();
15+
})
16+
.CreateLogger();
2617

27-
while (true)
28-
{
29-
Log.Information("Yo, RabbitMQ!");
30-
Thread.Sleep(1000);
31-
}
32-
}
33-
}
18+
while (true)
19+
{
20+
Log.Information("Yo, RabbitMQ!");
21+
Thread.Sleep(1000);
3422
}

‎seq-input-rabbitmq.sln.DotSettings‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
<wpf:ResourceDictionary xml:space="preserve" xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml" xmlns:s="clr-namespace:System;assembly=mscorlib" xmlns:ss="urn:shemas-jetbrains-com:settings-storage-xaml" xmlns:wpf="http://schemas.microsoft.com/winfx/2006/xaml/presentation">
2+
<s:String x:Key="/Default/CodeStyle/Naming/CSharpNaming/Abbreviations/=MQV/@EntryIndexedValue">MQV</s:String></wpf:ResourceDictionary>

‎src/Seq.Input.RabbitMQ/RabbitMQInput.cs‎

Lines changed: 125 additions & 118 deletions
Original file line numberDiff line numberDiff line change
@@ -1,134 +1,141 @@
11
using System;
22
using System.IO;
33
using System.Text;
4+
using System.Threading.Tasks;
45
using Seq.Apps;
6+
// ReSharper disable MemberCanBePrivate.Global, UnusedType.Global, UnusedAutoPropertyAccessor.Global
57

6-
namespace Seq.Input.RabbitMQ
8+
namespace Seq.Input.RabbitMQ;
9+
10+
[SeqApp("RabbitMQ Input",
11+
Description = "Pulls JSON-formatted events from a RabbitMQ queue. For details of the " +
12+
"supported JSON schema, see " +
13+
"https://github.com/serilog/serilog-formatting-compact/#format-details.")]
14+
public sealed class RabbitMQInput : SeqApp, IPublishJson, IDisposable
715
{
8-
[SeqApp("RabbitMQ Input",
9-
Description = "Pulls JSON-formatted events from a RabbitMQ queue. For details of the " +
10-
"supported JSON schema, see " +
11-
"https://github.com/serilog/serilog-formatting-compact/#format-details.")]
12-
public class RabbitMQInput : SeqApp, IPublishJson, IDisposable
16+
RabbitMQListener _listener;
17+
18+
[SeqAppSetting(
19+
DisplayName = "RabbitMQ host",
20+
IsOptional = true,
21+
HelpText = "The hostname on which RabbitMQ is running. The default is `localhost`.")]
22+
public string RabbitMQHost { get; set; } = "localhost";
23+
24+
[SeqAppSetting(
25+
DisplayName = "RabbitMQ Virtual Host",
26+
IsOptional = true,
27+
HelpText = "The virtual host in RabbitMQ. The default is `/`.")]
28+
public string RabbitMQVHost { get; set; } = "/";
29+
30+
[SeqAppSetting(
31+
DisplayName = "RabbitMQ port",
32+
IsOptional = true,
33+
HelpText = "The port on which the RabbitMQ server is listening. The default is `5672`.")]
34+
public int RabbitMQPort { get; set; } = 5672;
35+
36+
[SeqAppSetting(
37+
DisplayName = "RabbitMQ user",
38+
IsOptional = true,
39+
HelpText = "The username provided when connecting to RabbitMQ. The default is `guest`.")]
40+
public string RabbitMQUser { get; set; } = "guest";
41+
42+
[SeqAppSetting(
43+
DisplayName = "RabbitMQ password",
44+
IsOptional = true,
45+
InputType = SettingInputType.Password,
46+
HelpText = "The password provided when connecting to RabbitMQ. The default is `guest`.")]
47+
public string RabbitMQPassword { get; set; } = "guest";
48+
49+
[SeqAppSetting(
50+
DisplayName = "RabbitMQ queue",
51+
IsOptional = true,
52+
HelpText = "The RabbitMQ queue name to receive events from. The default is `Logs`.")]
53+
public string RabbitMQQueue { get; set; } = "logs";
54+
55+
[SeqAppSetting(
56+
DisplayName = "Require SSL",
57+
IsOptional = true,
58+
HelpText = "Whether or not the connection is with SSL. The default is false.")]
59+
public bool IsSsl { get; set; }
60+
61+
[SeqAppSetting(
62+
DisplayName = "Durable",
63+
IsOptional = true,
64+
HelpText = "Whether or not the queue is durable. The default is false.")]
65+
public bool IsQueueDurable { get; set; }
66+
67+
[SeqAppSetting(
68+
DisplayName = "Exclusive",
69+
IsOptional = true,
70+
HelpText = "Whether or not the queue is exclusive. The default is false.")]
71+
public bool IsQueueExclusive { get; set; }
72+
73+
[SeqAppSetting(
74+
DisplayName = "Auto-delete",
75+
IsOptional = true,
76+
HelpText = "Whether or not the queue subscription is durable. The default is false.")]
77+
public bool IsQueueAutoDelete { get; set; }
78+
79+
[SeqAppSetting(
80+
DisplayName = "Auto-ACK",
81+
IsOptional = true,
82+
HelpText = "Whether or not messages should be auto-acknowledged. The default is true.")]
83+
public bool IsReceiveAutoAck { get; set; } = true;
84+
85+
[SeqAppSetting(
86+
DisplayName = "Dead Letter Exchange",
87+
IsOptional = true,
88+
HelpText = "The name of the dead letter exchange associated with this queue. If specified, the exchange will be used when declaring the queue, otherwise no dead lettering will be configured.")]
89+
public string Dlx { get; set; }
90+
91+
public void Start(TextWriter inputWriter)
1392
{
14-
RabbitMQListener _listener;
15-
16-
[SeqAppSetting(
17-
DisplayName = "RabbitMQ host",
18-
IsOptional = true,
19-
HelpText = "The hostname on which RabbitMQ is running. The default is `localhost`.")]
20-
public string RabbitMQHost { get; set; } = "localhost";
21-
22-
[SeqAppSetting(
23-
DisplayName = "RabbitMQ Virtual Host",
24-
IsOptional = true,
25-
HelpText = "The virtual host in RabbitMQ. The default is `/`.")]
26-
public string RabbitMQVHost { get; set; } = "/";
27-
28-
[SeqAppSetting(
29-
DisplayName = "RabbitMQ port",
30-
IsOptional = true,
31-
HelpText = "The port on which the RabbitMQ server is listening. The default is `5672`.")]
32-
public int RabbitMQPort { get; set; } = 5672;
33-
34-
[SeqAppSetting(
35-
DisplayName = "RabbitMQ user",
36-
IsOptional = true,
37-
HelpText = "The username provided when connecting to RabbitMQ. The default is `guest`.")]
38-
public string RabbitMQUser { get; set; } = "guest";
39-
40-
[SeqAppSetting(
41-
DisplayName = "RabbitMQ password",
42-
IsOptional = true,
43-
InputType = SettingInputType.Password,
44-
HelpText = "The password provided when connecting to RabbitMQ. The default is `guest`.")]
45-
public string RabbitMQPassword { get; set; } = "guest";
46-
47-
[SeqAppSetting(
48-
DisplayName = "RabbitMQ queue",
49-
IsOptional = true,
50-
HelpText = "The RabbitMQ queue name to receive events from. The default is `Logs`.")]
51-
public string RabbitMQQueue { get; set; } = "logs";
52-
53-
[SeqAppSetting(
54-
DisplayName = "Require SSL",
55-
IsOptional = true,
56-
HelpText = "Whether or not the connection is with SSL. The default is false.")]
57-
public bool IsSsl { get; set; }
58-
59-
[SeqAppSetting(
60-
DisplayName = "Durable",
61-
IsOptional = true,
62-
HelpText = "Whether or not the queue is durable. The default is false.")]
63-
public bool IsQueueDurable { get; set; }
64-
65-
[SeqAppSetting(
66-
DisplayName = "Exclusive",
67-
IsOptional = true,
68-
HelpText = "Whether or not the queue is exclusive. The default is false.")]
69-
public bool IsQueueExclusive { get; set; }
70-
71-
[SeqAppSetting(
72-
DisplayName = "Auto-delete",
73-
IsOptional = true,
74-
HelpText = "Whether or not the queue subscription is durable. The default is false.")]
75-
public bool IsQueueAutoDelete { get; set; }
76-
77-
[SeqAppSetting(
78-
DisplayName = "Auto-ACK",
79-
IsOptional = true,
80-
HelpText = "Whether or not messages should be auto-acknowledged. The default is true.")]
81-
public bool IsReceiveAutoAck { get; set; } = true;
82-
83-
[SeqAppSetting(
84-
DisplayName = "Dead Letter Exchange",
85-
IsOptional = true,
86-
HelpText = "The name of the dead letter exchange associated with this queue. If specified, the exchange will be used when declaring the queue, otherwise no dead lettering will be configured.")]
87-
public string Dlx { get; set; }
88-
89-
public void Start(TextWriter inputWriter)
93+
var sync = new object();
94+
Task ReceiveAsync(ReadOnlyMemory<byte> body)
9095
{
91-
var sync = new object();
92-
void Receive(ReadOnlyMemory<byte> body)
96+
try
9397
{
94-
try
95-
{
96-
lock (sync)
97-
{
98-
var clef = Encoding.UTF8.GetString(body.ToArray());
99-
inputWriter.WriteLine(clef);
100-
}
101-
}
102-
catch (Exception ex)
98+
lock (sync)
10399
{
104-
Log.Error(ex, "A received message could not be decoded");
100+
var clef = Encoding.UTF8.GetString(body.ToArray());
101+
inputWriter.WriteLine(clef);
105102
}
106103
}
107-
108-
_listener = new RabbitMQListener(
109-
Receive,
110-
RabbitMQHost,
111-
RabbitMQVHost,
112-
RabbitMQPort,
113-
RabbitMQUser,
114-
RabbitMQPassword,
115-
RabbitMQQueue,
116-
IsSsl,
117-
IsQueueDurable,
118-
IsQueueAutoDelete,
119-
IsQueueExclusive,
120-
IsReceiveAutoAck,
121-
Dlx);
104+
catch (Exception ex)
105+
{
106+
Log.Error(ex, "A received message could not be decoded");
107+
}
108+
109+
return Task.CompletedTask;
122110
}
123111

124-
public void Stop()
125-
{
126-
_listener.Close();
127-
}
112+
// Not a deadlock risk on .NET 8, but ideally we'll introduce `IPublishJsonAsync` and provide
113+
// async start/stop/dispose variants.
114+
_listener = RabbitMQListener.CreateAsync(
115+
ReceiveAsync,
116+
RabbitMQHost,
117+
RabbitMQVHost,
118+
RabbitMQPort,
119+
RabbitMQUser,
120+
RabbitMQPassword,
121+
RabbitMQQueue,
122+
IsSsl,
123+
IsQueueDurable,
124+
IsQueueAutoDelete,
125+
IsQueueExclusive,
126+
IsReceiveAutoAck,
127+
Dlx).Result;
128+
}
128129

129-
public void Dispose()
130-
{
131-
_listener?.Dispose();
132-
}
130+
public void Stop()
131+
{
132+
// Not a deadlock risk on .NET 8, but ideally we'll introduce `IPublishJsonAsync` and provide
133+
// async start/stop/dispose variants.
134+
_listener.CloseAsync().Wait();
135+
}
136+
137+
public void Dispose()
138+
{
139+
_listener?.Dispose();
133140
}
134-
}
141+
}

0 commit comments

Comments
 (0)