Browse Source

Enable nullable annotations for Volo.Abp.Kafka

pull/17557/head
liangshiwei 3 years ago
parent
commit
438bd42359
  1. 2
      framework/src/Volo.Abp.Kafka/Volo.Abp.Kafka.csproj
  2. 6
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/AbpKafkaOptions.cs
  3. 2
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs
  4. 2
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/IConsumerPool.cs
  5. 2
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/IKafkaMessageConsumerFactory.cs
  6. 2
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/IProducerPool.cs
  7. 4
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaConnections.cs
  8. 12
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs
  9. 2
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumerFactory.cs
  10. 2
      framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs

2
framework/src/Volo.Abp.Kafka/Volo.Abp.Kafka.csproj

@ -5,6 +5,8 @@
<PropertyGroup> <PropertyGroup>
<TargetFrameworks>netstandard2.0;netstandard2.1;net7.0</TargetFrameworks> <TargetFrameworks>netstandard2.0;netstandard2.1;net7.0</TargetFrameworks>
<Nullable>enable</Nullable>
<WarningsAsErrors>Nullable</WarningsAsErrors>
<RootNamespace /> <RootNamespace />
</PropertyGroup> </PropertyGroup>

6
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/AbpKafkaOptions.cs

@ -8,11 +8,11 @@ public class AbpKafkaOptions
{ {
public KafkaConnections Connections { get; } public KafkaConnections Connections { get; }
public Action<ProducerConfig> ConfigureProducer { get; set; } public Action<ProducerConfig>? ConfigureProducer { get; set; }
public Action<ConsumerConfig> ConfigureConsumer { get; set; } public Action<ConsumerConfig>? ConfigureConsumer { get; set; }
public Action<TopicSpecification> ConfigureTopic { get; set; } public Action<TopicSpecification>? ConfigureTopic { get; set; }
public AbpKafkaOptions() public AbpKafkaOptions()
{ {

2
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ConsumerPool.cs

@ -30,7 +30,7 @@ public class ConsumerPool : IConsumerPool, ISingletonDependency
Logger = new NullLogger<ConsumerPool>(); Logger = new NullLogger<ConsumerPool>();
} }
public virtual IConsumer<string, byte[]> Get(string groupId, string connectionName = null) public virtual IConsumer<string, byte[]> Get(string groupId, string? connectionName = null)
{ {
connectionName ??= KafkaConnections.DefaultConnectionName; connectionName ??= KafkaConnections.DefaultConnectionName;

2
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/IConsumerPool.cs

@ -5,5 +5,5 @@ namespace Volo.Abp.Kafka;
public interface IConsumerPool : IDisposable public interface IConsumerPool : IDisposable
{ {
IConsumer<string, byte[]> Get(string groupId, string connectionName = null); IConsumer<string, byte[]> Get(string groupId, string? connectionName = null);
} }

2
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/IKafkaMessageConsumerFactory.cs

@ -14,5 +14,5 @@ public interface IKafkaMessageConsumerFactory
IKafkaMessageConsumer Create( IKafkaMessageConsumer Create(
string topicName, string topicName,
string groupId, string groupId,
string connectionName = null); string? connectionName = null);
} }

2
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/IProducerPool.cs

@ -5,5 +5,5 @@ namespace Volo.Abp.Kafka;
public interface IProducerPool : IDisposable public interface IProducerPool : IDisposable
{ {
IProducer<string, byte[]> Get(string connectionName = null); IProducer<string, byte[]> Get(string? connectionName = null);
} }

4
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaConnections.cs

@ -21,8 +21,10 @@ public class KafkaConnections : Dictionary<string, ClientConfig>
Default = new ClientConfig(); Default = new ClientConfig();
} }
public ClientConfig GetOrDefault(string connectionName) public ClientConfig GetOrDefault(string? connectionName)
{ {
connectionName ??= DefaultConnectionName;
if (TryGetValue(connectionName, out var connectionFactory)) if (TryGetValue(connectionName, out var connectionFactory))
{ {
return connectionFactory; return connectionFactory;

12
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumer.cs

@ -31,13 +31,13 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency,
protected ConcurrentBag<Func<Message<string, byte[]>, Task>> Callbacks { get; } protected ConcurrentBag<Func<Message<string, byte[]>, Task>> Callbacks { get; }
protected IConsumer<string, byte[]> Consumer { get; private set; } protected IConsumer<string, byte[]>? Consumer { get; private set; }
protected string ConnectionName { get; private set; } protected string? ConnectionName { get; private set; }
protected string GroupId { get; private set; } protected string GroupId { get; private set; } = default!;
protected string TopicName { get; private set; } protected string TopicName { get; private set; } = default!;
public KafkaMessageConsumer( public KafkaMessageConsumer(
IConsumerPool consumerPool, IConsumerPool consumerPool,
@ -63,7 +63,7 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency,
public virtual void Initialize( public virtual void Initialize(
[NotNull] string topicName, [NotNull] string topicName,
[NotNull] string groupId, [NotNull] string groupId,
string connectionName = null) string? connectionName = null)
{ {
Check.NotNull(topicName, nameof(topicName)); Check.NotNull(topicName, nameof(topicName));
Check.NotNull(groupId, nameof(groupId)); Check.NotNull(groupId, nameof(groupId));
@ -160,7 +160,7 @@ public class KafkaMessageConsumer : IKafkaMessageConsumer, ITransientDependency,
} }
finally finally
{ {
Consumer.Commit(consumeResult); Consumer?.Commit(consumeResult);
} }
} }

2
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/KafkaMessageConsumerFactory.cs

@ -16,7 +16,7 @@ public class KafkaMessageConsumerFactory : IKafkaMessageConsumerFactory, ISingle
public IKafkaMessageConsumer Create( public IKafkaMessageConsumer Create(
string topicName, string topicName,
string groupId, string groupId,
string connectionName = null) string? connectionName = null)
{ {
var consumer = ServiceScope.ServiceProvider.GetRequiredService<KafkaMessageConsumer>(); var consumer = ServiceScope.ServiceProvider.GetRequiredService<KafkaMessageConsumer>();
consumer.Initialize(topicName, groupId, connectionName); consumer.Initialize(topicName, groupId, connectionName);

2
framework/src/Volo.Abp.Kafka/Volo/Abp/Kafka/ProducerPool.cs

@ -32,7 +32,7 @@ public class ProducerPool : IProducerPool, ISingletonDependency
Logger = new NullLogger<ProducerPool>(); Logger = new NullLogger<ProducerPool>();
} }
public virtual IProducer<string, byte[]> Get(string connectionName = null) public virtual IProducer<string, byte[]> Get(string? connectionName = null)
{ {
connectionName ??= KafkaConnections.DefaultConnectionName; connectionName ??= KafkaConnections.DefaultConnectionName;

Loading…
Cancel
Save