Vooz/Universley.OrleansContrib.StreamsProvider.Redis
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ce fini

Browse Source
This commit is contained in:
Samson Amaugo 2024-06-26 18:02:24 +01:00
parent 9d74f05f79
commit f38027398c
7 changed files with 10 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Client/Program.cs
View File

@ -1,5 +1,4 @@
using Microsoft.CodeAnalysis.Operations;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Orleans.Configuration;
@ -46,7 +45,7 @@ var task = Task.Run(async () =>
{
logger.LogInformation("Sending number {Number}", num);
await stream.OnNextAsync(num++);
if (num == 20)
{
break;

5
Provider/RedisStreamAdapter.cs
View File

@ -2,12 +2,7 @@
using Orleans.Runtime;
using Orleans.Streams;
using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;
namespace Provider
{

5
Provider/RedisStreamBatchContainer.cs
View File

@ -1,12 +1,7 @@
using Orleans.Runtime;
using Orleans.Streams;
using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;
namespace Provider
{

3
Provider/RedisStreamFactory.cs
View File

@ -1,6 +1,5 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Orleans;
using Orleans.Configuration;
using Orleans.Providers.Streams.Common;
using Orleans.Streams;
@ -56,7 +55,7 @@ namespace Provider
public IQueueAdapterCache GetQueueAdapterCache()
{
return new SimpleQueueAdapterCache(_simpleQueueCacheOptions,_providerName, _loggerFactory);
return new SimpleQueueAdapterCache(_simpleQueueCacheOptions, _providerName, _loggerFactory);
}
public IStreamQueueMapper GetStreamQueueMapper()

7
Provider/RedisStreamFailureHandler.cs
View File

@ -1,11 +1,6 @@
using Microsoft.Extensions.Logging;
using Orleans.Runtime;
using Orleans.Streams;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Provider
{
@ -19,7 +14,7 @@ namespace Provider
}
public bool ShouldFaultSubsriptionOnError => true;
public Task OnDeliveryFailure(GuidId subscriptionId, string streamProviderName, StreamId streamIdentity, StreamSequenceToken sequenceToken)
{

16
Provider/RedisStreamSequenceToken.cs
View File

@ -1,11 +1,5 @@
using Orleans;
using Orleans.Streams;
using Orleans.Streams;
using StackExchange.Redis;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
namespace Provider
{
@ -15,7 +9,7 @@ namespace Provider
[Id(0)]
public override long SequenceNumber { get; protected set; }
[Id(1)]
public override int EventIndex { get ; protected set ; }
public override int EventIndex { get; protected set; }
public RedisStreamSequenceToken(RedisValue id)
{
@ -31,10 +25,10 @@ namespace Provider
}
public override int CompareTo(StreamSequenceToken other)
{
if(other is null) throw new ArgumentNullException(nameof(other));
if(other is RedisStreamSequenceToken token)
if (other is null) throw new ArgumentNullException(nameof(other));
if (other is RedisStreamSequenceToken token)
{
if(SequenceNumber == token.SequenceNumber)
if (SequenceNumber == token.SequenceNumber)
{
return EventIndex.CompareTo(token.EventIndex);
}

5
Server/Program.cs
View File

@ -1,12 +1,9 @@

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Orleans.Configuration;
using Orleans.Runtime;
using Orleans.Streams;
using Provider;
using StackExchange.Redis;
var builder = new HostBuilder()