Browse Source

Dispose channel pool on application shutdown.

pull/395/head
Halil ibrahim Kalkan 8 years ago
parent
commit
d70875311f
  1. 7
      framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpBackgroundJobsRabbitMqModule.cs
  2. 4
      framework/src/Volo.Abp.Core/Volo/Abp/Modularity/ModuleManager.cs
  3. 6
      framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqModule.cs
  4. 119
      framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs
  5. 16
      framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPoolItem.cs
  6. 2
      framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ConnectionPool.cs
  7. 6
      framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IChannelPool.cs

7
framework/src/Volo.Abp.BackgroundJobs.RabbitMQ/Volo/Abp/BackgroundJobs/RabbitMQ/AbpBackgroundJobsRabbitMqModule.cs

@ -25,5 +25,12 @@ namespace Volo.Abp.BackgroundJobs.RabbitMQ
.GetRequiredService<IJobQueueManager>() .GetRequiredService<IJobQueueManager>()
.Start(); .Start();
} }
public override void OnApplicationShutdown(ApplicationShutdownContext context)
{
context.ServiceProvider
.GetRequiredService<IJobQueueManager>()
.Stop();
}
} }
} }

4
framework/src/Volo.Abp.Core/Volo/Abp/Modularity/ModuleManager.cs

@ -57,9 +57,11 @@ namespace Volo.Abp.Modularity
public void ShutdownModules(ApplicationShutdownContext context) public void ShutdownModules(ApplicationShutdownContext context)
{ {
var modules = _moduleContainer.Modules.Reverse().ToList();
foreach (var contributer in _lifecycleContributers) foreach (var contributer in _lifecycleContributers)
{ {
foreach (var module in _moduleContainer.Modules) foreach (var module in modules)
{ {
contributer.Shutdown(context, module.Instance); contributer.Shutdown(context, module.Instance);
} }

6
framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/AbpRabbitMqModule.cs

@ -17,10 +17,12 @@ namespace Volo.Abp.RabbitMQ
public override void OnApplicationShutdown(ApplicationShutdownContext context) public override void OnApplicationShutdown(ApplicationShutdownContext context)
{ {
context.ServiceProvider context.ServiceProvider
.GetRequiredService<IConnectionPool>() .GetRequiredService<IChannelPool>()
.Dispose(); .Dispose();
//TODO: Dispose channel pool when it's implemented! context.ServiceProvider
.GetRequiredService<IConnectionPool>()
.Dispose();
} }
} }
} }

119
framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPool.cs

@ -11,7 +11,11 @@ namespace Volo.Abp.RabbitMQ
protected IConnectionPool ConnectionPool { get; } protected IConnectionPool ConnectionPool { get; }
protected ConcurrentDictionary<string, ChannelPoolItem> Channels { get; } protected ConcurrentDictionary<string, ChannelPoolItem> Channels { get; }
protected bool IsDisposed { get; private set; }
protected TimeSpan ChannelDisposeWaitDuration { get; set; } = TimeSpan.FromSeconds(15);
public ChannelPool(IConnectionPool connectionPool) public ChannelPool(IConnectionPool connectionPool)
{ {
ConnectionPool = connectionPool; ConnectionPool = connectionPool;
@ -20,40 +24,115 @@ namespace Volo.Abp.RabbitMQ
public virtual IChannelAccessor Acquire(string channelName = null) public virtual IChannelAccessor Acquire(string channelName = null)
{ {
CheckDisposed();
channelName = channelName ?? ""; channelName = channelName ?? "";
var poolItem = Channels.GetOrAdd(channelName, _ => new ChannelPoolItem var poolItem = Channels.GetOrAdd(
channelName,
_ => new ChannelPoolItem(CreateChannel(channelName))
);
poolItem.Acquire();
return new ChannelAccessor(
poolItem.Channel,
() => poolItem.Release()
);
}
protected virtual IModel CreateChannel(string channelName)
{
//TODO: How to determine the right connection name?
return ConnectionPool.Get().CreateModel();
}
protected void CheckDisposed()
{
if (IsDisposed)
{
throw new ObjectDisposedException(nameof(ChannelPool));
}
}
public void Dispose()
{
if (IsDisposed)
{ {
Channel = CreateChannel(channelName) return;
}); }
lock (poolItem) IsDisposed = true;
foreach (var poolItem in Channels.Values)
{ {
while (poolItem.IsInUse) try
{ {
Monitor.Wait(poolItem); poolItem.WaitIfInUse(ChannelDisposeWaitDuration);
poolItem.Channel.Dispose();
} }
catch
poolItem.IsInUse = true; { }
} }
return new ChannelAccessor( Channels.Clear();
poolItem.Channel, }
() =>
protected class ChannelPoolItem : IDisposable
{
public IModel Channel { get; }
public bool IsInUse
{
get => _isInUse;
private set => _isInUse = value;
}
private volatile bool _isInUse;
public ChannelPoolItem(IModel channel)
{
Channel = channel;
}
public void Acquire()
{
lock (this)
{ {
lock (poolItem) while (IsInUse)
{ {
poolItem.IsInUse = false; Monitor.Wait(this);
Monitor.PulseAll(poolItem);
} }
IsInUse = true;
} }
); }
}
protected virtual IModel CreateChannel(string channelName) public void WaitIfInUse(TimeSpan timeout)
{ {
//TODO: How to determine the right connection name? lock (this)
return ConnectionPool.Get().CreateModel(); {
if (!IsInUse)
{
return;
}
Monitor.Wait(this, timeout);
}
}
public void Release()
{
lock (this)
{
IsInUse = false;
Monitor.PulseAll(this);
}
}
public void Dispose()
{
Channel.Dispose();
}
} }
protected class ChannelAccessor : IChannelAccessor protected class ChannelAccessor : IChannelAccessor

16
framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ChannelPoolItem.cs

@ -1,16 +0,0 @@
using RabbitMQ.Client;
namespace Volo.Abp.RabbitMQ
{
public class ChannelPoolItem
{
public IModel Channel { get; set; }
public bool IsInUse
{
get => _isInUse;
set => _isInUse = value;
}
private volatile bool _isInUse;
}
}

2
framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/ConnectionPool.cs

@ -53,6 +53,8 @@ namespace Volo.Abp.RabbitMQ
} }
} }
Connections.Clear();
} }
} }
} }

6
framework/src/Volo.Abp.RabbitMQ/Volo/Abp/RabbitMQ/IChannelPool.cs

@ -1,6 +1,8 @@
namespace Volo.Abp.RabbitMQ using System;
namespace Volo.Abp.RabbitMQ
{ {
public interface IChannelPool public interface IChannelPool : IDisposable
{ {
IChannelAccessor Acquire(string channelName = null); IChannelAccessor Acquire(string channelName = null);
} }

Loading…
Cancel
Save