Skip to content

Commit 08ef26d

Browse files
author
Carlos Lisandra Osborne
committed
Add multiple targets.
Add comments. Add version.props.
1 parent fd8524b commit 08ef26d

29 files changed

+330
-32
lines changed

‎Bluekiri.Consumer.sln

+18-3
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11

22
Microsoft Visual Studio Solution File, Format Version 12.00
3-
# Visual Studio Version 16
4-
VisualStudioVersion = 16.0.29230.47
3+
# Visual Studio Version 17
4+
VisualStudioVersion = 17.0.32112.339
55
MinimumVisualStudioVersion = 15.0.26124.0
66
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "src", "src", "{F54E4B43-2123-4E84-948F-646FF336BCA3}"
77
EndProject
@@ -19,7 +19,9 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Bluekiri.Consumer.Tests", "
1919
EndProject
2020
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Bluekiri.Consumer.Kafka", "src\Kafka\src\Bluekiri.Consumer.Kafka\Bluekiri.Consumer.Kafka.csproj", "{B9783E86-0FD9-49DA-9C3D-365C3939C47F}"
2121
EndProject
22-
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Sample.Application.Handlers", "sample\Sample.Application.Handlers\Sample.Application.Handlers.csproj", "{4607D8F3-B373-46E9-A99B-F992AB42A671}"
22+
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sample.Application.Handlers", "sample\Sample.Application.Handlers\Sample.Application.Handlers.csproj", "{4607D8F3-B373-46E9-A99B-F992AB42A671}"
23+
EndProject
24+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Bluekiri.Consumer.Abstractions", "src\Bluekiri.Consumer.Abstractions\Bluekiri.Consumer.Abstractions.csproj", "{78E648B0-C786-4560-B70A-6C4D57112CAC}"
2325
EndProject
2426
Global
2527
GlobalSection(SolutionConfigurationPlatforms) = preSolution
@@ -91,6 +93,18 @@ Global
9193
{4607D8F3-B373-46E9-A99B-F992AB42A671}.Release|x64.Build.0 = Release|Any CPU
9294
{4607D8F3-B373-46E9-A99B-F992AB42A671}.Release|x86.ActiveCfg = Release|Any CPU
9395
{4607D8F3-B373-46E9-A99B-F992AB42A671}.Release|x86.Build.0 = Release|Any CPU
96+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
97+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Debug|Any CPU.Build.0 = Debug|Any CPU
98+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Debug|x64.ActiveCfg = Debug|Any CPU
99+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Debug|x64.Build.0 = Debug|Any CPU
100+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Debug|x86.ActiveCfg = Debug|Any CPU
101+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Debug|x86.Build.0 = Debug|Any CPU
102+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Release|Any CPU.ActiveCfg = Release|Any CPU
103+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Release|Any CPU.Build.0 = Release|Any CPU
104+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Release|x64.ActiveCfg = Release|Any CPU
105+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Release|x64.Build.0 = Release|Any CPU
106+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Release|x86.ActiveCfg = Release|Any CPU
107+
{78E648B0-C786-4560-B70A-6C4D57112CAC}.Release|x86.Build.0 = Release|Any CPU
94108
EndGlobalSection
95109
GlobalSection(SolutionProperties) = preSolution
96110
HideSolutionNode = FALSE
@@ -102,6 +116,7 @@ Global
102116
{880B79AF-8F33-4FCD-A748-5040D6AD9ACB} = {0BE61F1D-882E-4A3A-8705-17872C1F670D}
103117
{B9783E86-0FD9-49DA-9C3D-365C3939C47F} = {F54E4B43-2123-4E84-948F-646FF336BCA3}
104118
{4607D8F3-B373-46E9-A99B-F992AB42A671} = {E8A4DB02-4153-43FC-9004-74129A20A957}
119+
{78E648B0-C786-4560-B70A-6C4D57112CAC} = {F54E4B43-2123-4E84-948F-646FF336BCA3}
105120
EndGlobalSection
106121
GlobalSection(ExtensibilityGlobals) = postSolution
107122
SolutionGuid = {544560F1-0479-4BFA-8243-E45D885B689C}

‎Directory.Build.Targets

+1
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
<Project/>

‎images/logo.png

4.35 KB
Loading

‎sample/Sample.Application/Program.cs

+1-1
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ public static async Task Main(string[] args)
3131
services.AddSingleton(typeof(SampleHandler));
3232
services.AddConsumerConfiguration<KafkaConsumer, KafkaConsumerOptions>(o =>
3333
{
34-
o.Topics.Add("test-topic");
34+
o.AddTopic("test-topic");
3535
o.SetProperty("bootstrap.servers", "lgmadanydkfk02v.corp.logitravelgroup.com:9092");
3636
o.SetProperty("group.id", "test_1");
3737
o.SetProperty("enable.auto.commit", "false");

‎src/Bluekiri.Consumer.Abstractions/Bluekiri.Consumer.Abstractions.csproj

+13
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<TargetFrameworks>netstandard2.0;netcoreapp3.1;net6.0</TargetFrameworks>
5+
<AssemblyName>Bluekiri.Consumer.Abstractions</AssemblyName>
6+
<RootNamespace>Bluekiri.Consumer.Abstractions</RootNamespace>
7+
<Company>Bluekiri</Company>
8+
<PackageIcon>images\logo.png</PackageIcon>
9+
</PropertyGroup>
10+
<ItemGroup>
11+
<None Include="..\..\images\logo.png" Pack="true" Visible="false" PackagePath="images\logo.png"/>
12+
</ItemGroup>
13+
</Project>

‎src/Bluekiri.Consumer.Abstractions/IBrokerConsumer.cs

+22
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
using System;
2+
using System.Threading;
3+
4+
namespace Bluekiri.Consumer.Abstractions
5+
{
6+
/// <summary>
7+
/// Abstraction of consumer.
8+
/// </summary>
9+
public interface IBrokerConsumer : IDisposable
10+
{
11+
/// <summary>
12+
/// Consume a message from a broker.
13+
/// </summary>
14+
/// <param name="stopingToken"><see cref="CancellationToken"/></param>
15+
/// <returns>A <see cref="MessageInfo"/></returns>
16+
MessageInfo Consume(CancellationToken stopingToken);
17+
/// <summary>
18+
/// Return true if your implementation allows auto commit.
19+
/// </summary>
20+
bool IsEnabledAutoCommit { get; }
21+
}
22+
}

‎src/Bluekiri.Consumer.Abstractions/IMessage.cs

+4
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
namespace Bluekiri.Consumer.Abstractions
2+
{
3+
public interface IMessage { }
4+
}

‎src/Bluekiri.Consumer.Abstractions/IMessageHandler.cs

+21
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
using System.Threading;
2+
using System.Threading.Tasks;
3+
4+
namespace Bluekiri.Consumer.Abstractions
5+
{
6+
/// <summary>
7+
/// Handler interface.
8+
/// </summary>
9+
/// <typeparam name="TMessage"><see cref="IMessage"/></typeparam>
10+
public interface IMessageHandler<in TMessage>
11+
where TMessage : IMessage
12+
{
13+
/// <summary>
14+
/// Handle Method.
15+
/// </summary>
16+
/// <param name="message"></param>
17+
/// <param name="cancellationToken"><see cref="CancellationToken"/></param>
18+
/// <returns></returns>
19+
Task Handle(TMessage message, CancellationToken cancellationToken = default);
20+
}
21+
}

‎src/Bluekiri.Consumer.Abstractions/MessageInfo.cs

+34
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
using System.Collections.Generic;
2+
using System.Threading.Tasks;
3+
4+
namespace Bluekiri.Consumer.Abstractions
5+
{
6+
/// <summary>
7+
/// MessageInfo
8+
/// </summary>
9+
public abstract class MessageInfo
10+
{
11+
public virtual string Key { get; set; }
12+
/// <summary>
13+
/// Message Header
14+
/// </summary>
15+
public virtual IDictionary<string, byte[]> Headers { get; set; }
16+
/// <summary>
17+
/// Message Value
18+
/// </summary>
19+
public virtual byte[] Message { get; set; }
20+
21+
/// <summary>
22+
/// Constructor.
23+
/// </summary>
24+
protected MessageInfo()
25+
{
26+
Headers = new Dictionary<string, byte[]>();
27+
}
28+
/// <summary>
29+
/// Commit if it is necessary the message.
30+
/// </summary>
31+
/// <returns></returns>
32+
public abstract Task CommitAsync();
33+
}
34+
}

‎src/Bluekiri.Consumer.Abstractions/MessageTypeAttribute.cs

+41
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
using System;
2+
3+
namespace Bluekiri.Consumer.Abstractions
4+
{
5+
/// <summary>
6+
/// Attribute used for detect handlers
7+
/// </summary>
8+
[AttributeUsage(AttributeTargets.Class, AllowMultiple = false)]
9+
public class MessageTypeAttribute : Attribute
10+
{
11+
12+
private readonly string _name;
13+
private readonly Type _modelType;
14+
/// <summary>
15+
/// Constructor.
16+
/// </summary>
17+
/// <param name="name">Identifier name</param>
18+
/// <param name="type">Model type</param>
19+
public MessageTypeAttribute(string name, Type type)
20+
{
21+
22+
this._name = name;
23+
this._modelType = type;
24+
25+
}
26+
/// <summary>
27+
/// Returns name.
28+
/// </summary>
29+
public virtual string Name
30+
{
31+
get { return _name; }
32+
}
33+
/// <summary>
34+
/// Returns model type.
35+
/// </summary>
36+
public virtual Type ModelType
37+
{
38+
get { return _modelType; }
39+
}
40+
}
41+
}

‎src/Bluekiri.Consumer.Models/Bluekiri.Consumer.Models.csproj

+14
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<TargetFrameworks>netstandard2.0;netcoreapp3.1;net6.0</TargetFrameworks>
5+
<AssemblyName>Bluekiri.Consumer.Models</AssemblyName>
6+
<RootNamespace>Bluekiri.Consumer.Models</RootNamespace>
7+
<Company>Bluekiri</Company>
8+
<PackageIcon>images\logo.png</PackageIcon>
9+
</PropertyGroup>
10+
<ItemGroup>
11+
<None Include="..\..\images\logo.png" Pack="true" Visible="false" PackagePath="images\logo.png" />
12+
</ItemGroup>
13+
14+
</Project>

‎src/Bluekiri.Consumer.Models/Class1.cs

+7
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
namespace Bluekiri.Consumer.Models
2+
{
3+
public class Class1
4+
{
5+
6+
}
7+
}

‎src/Bluekiri.Consumer/Bluekiri.Consumer.csproj

+7-3
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,17 @@
11
<Project Sdk="Microsoft.NET.Sdk">
22

33
<PropertyGroup>
4-
<TargetFramework>netstandard2.0</TargetFramework>
4+
<TargetFrameworks>netstandard2.0;netcoreapp3.1;net6.0</TargetFrameworks>
55
<AssemblyName>Bluekiri.Consumer</AssemblyName>
66
<RootNamespace>Bluekiri.Consumer</RootNamespace>
7-
<PackageIconUrl>https://bluekiri.com/images/ms-icon-70x70.png</PackageIconUrl>
8-
<VersionPrefix>0.0.1</VersionPrefix>
97
<Company>Bluekiri</Company>
8+
<PackageIcon>images\logo.png</PackageIcon>
109
</PropertyGroup>
10+
11+
<ItemGroup>
12+
<None Include="..\..\images\logo.png" Pack="true" Visible="false" PackagePath="images\logo.png"/>
13+
</ItemGroup>
14+
1115
<ItemGroup>
1216
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="2.2.0" />
1317
<PackageReference Include="Microsoft.Extensions.Options" Version="2.2.0" />

‎src/Bluekiri.Consumer/ConsumerOptions.cs

+12-1
Original file line numberDiff line numberDiff line change
@@ -3,18 +3,29 @@
33

44
namespace Bluekiri.Consumer
55
{
6+
/// <summary>
7+
/// Consumer options configuration.
8+
/// </summary>
69
public class ConsumerOptions
710
{
811
internal IList<Type> MessageFormatters { get; }
12+
913

14+
/// <summary>
15+
/// Constructor.
16+
/// Adds Default JsonMessageFormatter
17+
/// </summary>
1018
public ConsumerOptions()
1119
{
1220
MessageFormatters = new List<Type>
1321
{
1422
typeof(JsonMessageFormatter)
1523
};
1624
}
17-
25+
/// <summary>
26+
/// Adds new formatter to list.
27+
/// </summary>
28+
/// <typeparam name="T"><see cref="IMessageFormatter"/></typeparam>
1829
public void AddMessageFormatter<T>() where T : IMessageFormatter
1930
{
2031
MessageFormatters.Add(typeof(T));

‎src/Bluekiri.Consumer/ConsumerService.cs

+38-9
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,10 @@
1212

1313
namespace Bluekiri.Consumer
1414
{
15+
/// <summary>
16+
/// Worker for consume.
17+
/// </summary>
18+
/// <typeparam name="TConsumerOptions"><see cref="ConsumerOptions"/></typeparam>
1519
public class ConsumerService<TConsumerOptions> : BackgroundService where TConsumerOptions : ConsumerOptions, new()
1620
{
1721
private readonly IBrokerConsumer _consumer;
@@ -20,6 +24,17 @@ namespace Bluekiri.Consumer
2024
private readonly IEnumerable<IMessageFormatter> _formatters;
2125
private readonly ILogger _logger;
2226

27+
private const string ContentType = "ContentType";
28+
private const string MessageType = "MessageType";
29+
30+
/// <summary>
31+
/// Constructor.
32+
/// </summary>
33+
/// <param name="consumer"><see cref="IBrokerConsumer"/></param>
34+
/// <param name="handlerManager"><see cref="IHandlerManager"/></param>
35+
/// <param name="factory"><see cref="IHandlerMessageFactory"/></param>
36+
/// <param name="formatters"><see cref="IEnumerable{IMessageFormatter}"/></param>
37+
/// <param name="logger"><see cref="ILogger{ConsumerService}"/></param>
2338
public ConsumerService(
2439
IBrokerConsumer consumer,
2540
IHandlerManager handlerManager,
@@ -33,7 +48,10 @@ public ConsumerService(
3348
_formatters = formatters;
3449
_logger = logger;
3550
}
36-
51+
/// <summary>
52+
/// Backgroundservice override.
53+
/// </summary>
54+
/// <param name="stoppingToken"><see cref="CancellationToken"/></param>
3755
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
3856
{
3957
while (!stoppingToken.IsCancellationRequested)
@@ -42,10 +60,19 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
4260
{
4361
var consumeResult = _consumer.Consume(stoppingToken);
4462
if (consumeResult is null) continue;
63+
if (!consumeResult.Headers.ContainsKey(ContentType))
64+
{
65+
_logger.LogError("Headers not contains key ContentType");
66+
continue;
67+
}
68+
if (!consumeResult.Headers.ContainsKey(MessageType))
69+
{
70+
_logger.LogError("Headers not contains key MessageType");
71+
continue;
72+
}
4573

46-
47-
var contentType = Encoding.UTF8.GetString(consumeResult.Headers["ContentType"]);
48-
var messageType = Encoding.UTF8.GetString(consumeResult.Headers["MessageType"]);
74+
var contentType = Encoding.UTF8.GetString(consumeResult.Headers[ContentType]);
75+
var messageType = Encoding.UTF8.GetString(consumeResult.Headers[MessageType]);
4976

5077
if (string.IsNullOrWhiteSpace(contentType))
5178
{
@@ -59,7 +86,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
5986
continue;
6087
}
6188

62-
var formatter = _formatters.FirstOrDefault(f => f.ContentType.Equals(contentType));
89+
var formatter = _formatters.Where(f => f.ContentType.Equals(contentType)).FirstOrDefault();
6390

6491
if (formatter is null)
6592
{
@@ -73,9 +100,9 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
73100
continue;
74101
}
75102
var result = formatter.Deserialize(consumeResult.Message, modelType);
76-
await _factory.Execute(result, stoppingToken).ConfigureAwait(false);
103+
await _factory.ExecuteAsync(result, stoppingToken).ConfigureAwait(false);
104+
77105

78-
//await handler.HandleAsync(result);
79106
if (!_consumer.IsEnabledAutoCommit)
80107
{
81108
await consumeResult.CommitAsync().ConfigureAwait(false);
@@ -88,10 +115,12 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
88115
}
89116
}
90117

91-
118+
/// <summary>
119+
/// base dispose
120+
/// </summary>
92121
public override void Dispose()
93122
{
94-
// _consumer?.Dispose();
123+
_consumer?.Dispose();
95124
base.Dispose();
96125
}
97126
}

0 commit comments

Comments
 (0)