Skip to content
  • P
    Projects
  • G
    Groups
  • S
    Snippets
  • Help

丁松杰 / Pole

  • This project
    • Loading...
  • Sign in
Go to a project
  • Project
  • Repository
  • Issues 0
  • Merge Requests 0
  • Pipelines
  • Wiki
  • Snippets
  • Members
  • Activity
  • Graph
  • Charts
  • Create a new issue
  • Jobs
  • Commits
  • Issue Boards
  • Files
  • Commits
  • Branches
  • Tags
  • Contributors
  • Graph
  • Compare
  • Charts
Commit 5de72aaf authored 5 years ago by 丁松杰's avatar 丁松杰
Browse files
Options
  • Browse Files
  • Download
  • Email Patches
  • Plain Diff

完成 清理时间等的配置

parent f78114ce master … v1.0.0
Show whitespace changes
Inline Side-by-side
Showing with 7 additions and 1 deletions
  • src/Pole.Core/PoleOptions.cs
  • src/Pole.Core/Processor/PendingMessageRetryProcessor.cs
  • src/Pole.Core/UnitOfWork/UnitOfWork.cs
src/Pole.Core/PoleOptions.cs
View file @ 5de72aaf
......@@ -11,6 +11,7 @@ namespace Pole.Core
public int ExpiredEventsPreBulkDeleteDelaySeconds { get; set; } = 3;
public int ExpiredEventsCollectIntervalSeconds { get; set; } = 60 * 60;
public int PublishedEventsExpiredAfterSeconds { get; set; } = 60 * 60;
public IServiceCollection Services { get; private set; }
}
}
This diff is collapsed. Click to expand it.
src/Pole.Core/Processor/PendingMessageRetryProcessor.cs
View file @ 5de72aaf
......@@ -72,6 +72,7 @@ namespace Pole.Core.Processor
pendingMessage.Retries++;
await producer.Publish(bytes);
pendingMessage.StatusName = nameof(EventStatus.Published);
pendingMessage.ExpiresAt = DateTime.UtcNow.AddSeconds(options.PublishedEventsExpiredAfterSeconds);
}
await eventStorage.BulkChangePublishStateAsync(pendingMessages);
}
......
This diff is collapsed. Click to expand it.
src/Pole.Core/UnitOfWork/UnitOfWork.cs
View file @ 5de72aaf
......@@ -12,6 +12,7 @@ using Pole.Core.Abstraction;
using Pole.Core.Serialization;
using Pole.Core.EventBus.Event;
using Pole.Core.EventBus.EventStorage;
using Microsoft.Extensions.Options;
namespace Pole.Core.UnitOfWork
{
......@@ -21,13 +22,15 @@ namespace Pole.Core.UnitOfWork
private readonly IEventTypeFinder eventTypeFinder;
private readonly ISerializer serializer;
private readonly IEventStorage eventStorage;
private readonly PoleOptions options;
private IBus bus;
public UnitOfWork(IProducer producer, IEventTypeFinder eventTypeFinder, ISerializer serializer, IEventStorage eventStorage)
public UnitOfWork(IProducer producer, IEventTypeFinder eventTypeFinder, ISerializer serializer, IEventStorage eventStorage, IOptions<PoleOptions> options)
{
this.producer = producer;
this.eventTypeFinder = eventTypeFinder;
this.serializer = serializer;
this.eventStorage = eventStorage;
this.options = options.Value;
}
public async Task CompeleteAsync(CancellationToken cancellationToken = default)
......@@ -44,6 +47,7 @@ namespace Pole.Core.UnitOfWork
var bytes = bytesTransport.GetBytes();
await producer.Publish(bytes);
@event.StatusName = nameof(EventStatus.Published);
@event.ExpiresAt = DateTime.UtcNow.AddSeconds(options.PublishedEventsExpiredAfterSeconds);
});
await eventStorage.BulkChangePublishStateAsync(bufferedEvents);
}
......
This diff is collapsed. Click to expand it.
  • Write
  • Preview
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment