Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus 클래스 참조

더 자세히 ...

Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus에 대한 상속 다이어그램 :

Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus에 대한 협력 다이어그램:

Public 멤버 함수
	RabbitMqMessageBus (RabbitMqMessageBusOptions options)
	RabbitMqMessageBus (RabbitMqMessageBusOptions options, IRabbitMqConnectionFactory connectionFactory)
Task	ConnectAsync (CancellationToken cancellationToken=default)
Task	PublishAsync (MessageEnvelope message, CancellationToken cancellationToken=default)
Task	SubscribeAsync (string route, Func< MessageEnvelope, CancellationToken, Task > handler, CancellationToken cancellationToken=default)
Task	DisconnectAsync (CancellationToken cancellationToken=default)
async ValueTask	DisposeAsync ()

속성
ConnectionState	State = ConnectionState.Disconnected [get, private set]
TransportKind	Kind [get]

Private 멤버 함수
async Task	HandleReceivedAsync (RabbitMqDelivery delivery, CancellationToken cancellationToken)
void	DeclareTopology (IRabbitMqChannel channel)
void	Cleanup ()

정적 Private 멤버 함수
static void	ValidateOptions (RabbitMqMessageBusOptions options)

Private 속성
readonly RabbitMqMessageBusOptions	_options
readonly IRabbitMqConnectionFactory	_connectionFactory
readonly Dictionary< string, List< Func< MessageEnvelope, CancellationToken, Task > > >	_handlers = new()
readonly object	_syncRoot = new()
IRabbitMqConnection?	_connection
IRabbitMqChannel?	_channel
string?	_consumerTag

상세한 설명

RabbitMQ Exchange, Queue 및 라우팅 키를 사용하는 메시지 버스입니다.

메시지 봉투는 JSON으로 직렬화되어 RabbitMQ 본문에 저장됩니다.

RabbitMqMessageBus.cs 파일의 26 번째 라인에서 정의되었습니다.

생성자 & 소멸자 문서화

RabbitMqMessageBus() [1/2]

Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.RabbitMqMessageBus ( RabbitMqMessageBusOptions options )

inline

기본 연결 팩토리와 지정한 설정으로 메시지 버스를 초기화합니다.

매개변수

options

RabbitMQ 연결 및 토폴로지 설정입니다.

RabbitMqMessageBus.cs 파일의 109 번째 라인에서 정의되었습니다.

        : this(options, new RabbitMqClientConnectionFactory())
    {
    }

RabbitMqMessageBus() [2/2]

Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.RabbitMqMessageBus ( RabbitMqMessageBusOptions options, IRabbitMqConnectionFactory connectionFactory )

inline

설정과 사용자 지정 연결 팩토리로 메시지 버스를 초기화합니다.

매개변수

options	RabbitMQ 연결 및 토폴로지 설정입니다.
connectionFactory	브로커 연결을 생성할 팩토리입니다.

예외

ArgumentNullException

필수 입력 인자 중 하나가 null인 경우 발생합니다.

RabbitMqMessageBus.cs 파일의 146 번째 라인에서 정의되었습니다.

    {
        _options = options ?? throw new ArgumentNullException(nameof(options));
        _connectionFactory = connectionFactory ?? throw new ArgumentNullException(nameof(connectionFactory));
        ValidateOptions(_options);
    }

다음을 참조함 : _connectionFactory, _options, ValidateOptions().

이 함수 내부에서 호출하는 함수들에 대한 그래프입니다.:

멤버 함수 문서화

Cleanup()

void Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.Cleanup ( )

inlineprivate

처리기와 소비자를 정리하고 채널 및 연결을 안전하게 닫아 해제합니다.

RabbitMqMessageBus.cs 파일의 611 번째 라인에서 정의되었습니다.

    {
        lock (_syncRoot)
        {
            _handlers.Clear();
        }
 
        if (_channel is not null)
        {
            try
            {
                if (!string.IsNullOrWhiteSpace(_consumerTag))
                {
                    _channel.BasicCancel(_consumerTag);
                }
            }
            catch
            {
                // Ignore consumer cancel failure during cleanup.
            }
 
            try
            {
                if (_channel.IsOpen)
                {
                    _channel.Close();
                }
            }
            catch
            {
                // Ignore channel close failure during cleanup.
            }
 
            _channel.Dispose();
            _channel = null;
        }
 
        if (_connection is not null)
        {
            try
            {
                if (_connection.IsOpen)
                {
                    _connection.Close();
                }
            }
            catch
            {
                // Ignore connection close failure during cleanup.
            }
 
            _connection.Dispose();
            _connection = null;
        }
 
        _consumerTag = null;
    }

다음을 참조함 : _channel, _connection, _consumerTag, _handlers, _syncRoot.

다음에 의해서 참조됨 : ConnectAsync(), DisconnectAsync().

이 함수를 호출하는 함수들에 대한 그래프입니다.:

ConnectAsync()

Task Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.ConnectAsync ( CancellationToken cancellationToken = default )

inline

브로커 연결과 채널을 생성하고 구성된 토폴로지를 선언합니다.

매개변수

cancellationToken

연결 취소 요청을 감시하는 토큰입니다.

반환값

비동기 연결 작업입니다.

RabbitMqMessageBus.cs 파일의 199 번째 라인에서 정의되었습니다.

    {
        cancellationToken.ThrowIfCancellationRequested();
 
        if (State == ConnectionState.Connected)
        {
            return Task.CompletedTask;
        }
 
        State = ConnectionState.Connecting;
 
        try
        {
            _connection = _connectionFactory.CreateConnection(_options);
            _channel = _connection.CreateChannel();
 
            DeclareTopology(_channel);
 
            State = ConnectionState.Connected;
            return Task.CompletedTask;
        }
        catch
        {
            State = ConnectionState.Faulted;
            Cleanup();
            throw;
        }
    }

다음을 참조함 : _channel, _connection, _connectionFactory, _options, Cleanup(), DeclareTopology(), State.

이 함수 내부에서 호출하는 함수들에 대한 그래프입니다.:

DeclareTopology()

void Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.DeclareTopology ( IRabbitMqChannel channel )

inlineprivate

구성된 Exchange와 Queue를 선언하고 기본 라우팅 키로 바인딩합니다.

매개변수

channel

토폴로지를 선언할 열린 채널입니다.

RabbitMqMessageBus.cs 파일의 583 번째 라인에서 정의되었습니다.

    {
        channel.ExchangeDeclare(
            exchange: _options.ExchangeName,
            type: _options.ExchangeType,
            durable: _options.Durable,
            autoDelete: _options.AutoDelete);
 
        channel.QueueDeclare(
            queue: _options.QueueName,
            durable: _options.Durable,
            exclusive: _options.Exclusive,
            autoDelete: _options.AutoDelete);
 
        channel.QueueBind(
            queue: _options.QueueName,
            exchange: _options.ExchangeName,
            routingKey: _options.RoutingKey);
    }

다음을 참조함 : _options, Dreamine.Communication.RabbitMQ.Infrastructure.IRabbitMqChannel.ExchangeDeclare(), Dreamine.Communication.RabbitMQ.Infrastructure.IRabbitMqChannel.QueueBind(), Dreamine.Communication.RabbitMQ.Infrastructure.IRabbitMqChannel.QueueDeclare().

다음에 의해서 참조됨 : ConnectAsync().

이 함수 내부에서 호출하는 함수들에 대한 그래프입니다.:

이 함수를 호출하는 함수들에 대한 그래프입니다.:

DisconnectAsync()

Task Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.DisconnectAsync ( CancellationToken cancellationToken = default )

inline

소비자, 채널 및 브로커 연결을 정리합니다.

매개변수

cancellationToken

연결 해제 취소 요청을 감시하는 토큰입니다.

반환값

비동기 연결 해제 작업입니다.

RabbitMqMessageBus.cs 파일의 418 번째 라인에서 정의되었습니다.

    {
        cancellationToken.ThrowIfCancellationRequested();
 
        State = ConnectionState.Disconnecting;
 
        try
        {
            Cleanup();
            State = ConnectionState.Disconnected;
            return Task.CompletedTask;
        }
        catch
        {
            State = ConnectionState.Faulted;
            throw;
        }
    }

다음을 참조함 : Cleanup(), State.

다음에 의해서 참조됨 : DisposeAsync().

이 함수 내부에서 호출하는 함수들에 대한 그래프입니다.:

이 함수를 호출하는 함수들에 대한 그래프입니다.:

DisposeAsync()

async ValueTask Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.DisposeAsync ( )

inline

RabbitMQ 메시지 버스 연결과 관련 리소스를 비동기적으로 해제합니다.

반환값

비동기 리소스 해제 작업입니다.

RabbitMqMessageBus.cs 파일의 453 번째 라인에서 정의되었습니다.

    {
        await DisconnectAsync().ConfigureAwait(false);
    }

다음을 참조함 : DisconnectAsync().

이 함수 내부에서 호출하는 함수들에 대한 그래프입니다.:

HandleReceivedAsync()

async Task Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.HandleReceivedAsync ( RabbitMqDelivery delivery, CancellationToken cancellationToken )

inlineprivate

브로커 배달 본문을 메시지로 역직렬화하고 라우트 처리기를 실행한 뒤 성공 또는 실패 확인을 전송합니다.

매개변수

delivery	처리할 RabbitMQ 배달 정보입니다.
cancellationToken	처리기 실행 취소 토큰입니다.

반환값

비동기 배달 처리 작업입니다.

예외

JsonException

배달 본문이 올바른 메시지 JSON이 아닌 경우 발생할 수 있습니다.

RabbitMqMessageBus.cs 파일의 498 번째 라인에서 정의되었습니다.

    {
        if (_channel is null)
        {
            return;
        }
 
        try
        {
            var message = JsonSerializer.Deserialize<MessageEnvelope>(delivery.Body.Span);
 
            if (message is null)
            {
                _channel.BasicReject(delivery.DeliveryTag, requeue: false);
                return;
            }
 
            var route = string.IsNullOrWhiteSpace(message.Route)
                ? delivery.RoutingKey
                : message.Route;
 
            List<Func<MessageEnvelope, CancellationToken, Task>> handlers;
 
            lock (_syncRoot)
            {
                if (!_handlers.TryGetValue(route, out var routeHandlers) &&
                    !_handlers.TryGetValue(delivery.RoutingKey, out routeHandlers))
                {
                    handlers = [];
                }
                else
                {
                    handlers = new List<Func<MessageEnvelope, CancellationToken, Task>>(routeHandlers);
                }
            }
 
            if (handlers.Count == 0)
            {
                _channel.BasicAck(delivery.DeliveryTag, multiple: false);
                return;
            }
 
            foreach (var handler in handlers)
            {
                await handler(message, cancellationToken).ConfigureAwait(false);
            }
 
            _channel.BasicAck(delivery.DeliveryTag, multiple: false);
        }
        catch
        {
            try
            {
                _channel.BasicNack(
                    delivery.DeliveryTag,
                    multiple: false,
                    requeue: _options.RequeueOnHandlerError);
            }
            catch
            {
                // Ignore acknowledgement failure during shutdown or channel fault.
            }
 
            throw;
        }
    }

다음을 참조함 : _channel, _handlers, _options, _syncRoot, Dreamine.Communication.RabbitMQ.Infrastructure.RabbitMqDelivery.Body, Dreamine.Communication.RabbitMQ.Infrastructure.RabbitMqDelivery.DeliveryTag, Dreamine.Communication.RabbitMQ.Infrastructure.RabbitMqDelivery.RoutingKey.

다음에 의해서 참조됨 : SubscribeAsync().

이 함수를 호출하는 함수들에 대한 그래프입니다.:

PublishAsync()

Task Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.PublishAsync ( MessageEnvelope message, CancellationToken cancellationToken = default )

inline

메시지를 JSON으로 직렬화하여 구성된 Exchange와 라우팅 키로 발행합니다.

매개변수

message	발행할 메시지입니다.
cancellationToken	발행 취소 요청을 감시하는 토큰입니다.

반환값

비동기 발행 작업입니다.

예외

InvalidOperationException

현재 객체 상태에서 Publish Async 작업을 수행할 수 없는 경우 발생합니다.

RabbitMqMessageBus.cs 파일의 268 번째 라인에서 정의되었습니다.

    {
        ArgumentNullException.ThrowIfNull(message);
        cancellationToken.ThrowIfCancellationRequested();
 
        if (_channel is null || State != ConnectionState.Connected)
        {
            throw new InvalidOperationException("RabbitMQ message bus is not connected.");
        }
 
        var route = string.IsNullOrWhiteSpace(message.Route)
            ? _options.RoutingKey
            : message.Route;
 
        var body = JsonSerializer.SerializeToUtf8Bytes(message);
 
        var properties = _channel.CreateBasicProperties();
        properties.Persistent = _options.PersistentMessages;
        properties.ContentType = "application/json";
        properties.Type = nameof(MessageEnvelope);
 
        _channel.BasicPublish(
            exchange: _options.ExchangeName,
            routingKey: route,
            mandatory: false,
            properties: properties,
            body: body);
 
        return Task.CompletedTask;
    }

다음을 참조함 : _channel, _options, State.

SubscribeAsync()

Task Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.SubscribeAsync ( string route, Func< MessageEnvelope, CancellationToken, Task > handler, CancellationToken cancellationToken = default )

inline

라우팅 키를 Queue에 바인딩하고 비동기 처리기를 등록합니다.

매개변수

route	구독할 라우팅 키입니다.
handler	수신 메시지 처리기입니다.
cancellationToken	구독 취소 요청을 감시하는 토큰입니다.

반환값

비동기 구독 작업입니다.

예외

InvalidOperationException

현재 객체 상태에서 Subscribe Async 작업을 수행할 수 없는 경우 발생합니다.

RabbitMqMessageBus.cs 파일의 349 번째 라인에서 정의되었습니다.

    {
        ArgumentException.ThrowIfNullOrWhiteSpace(route);
        ArgumentNullException.ThrowIfNull(handler);
        cancellationToken.ThrowIfCancellationRequested();
 
        if (_channel is null || State != ConnectionState.Connected)
        {
            throw new InvalidOperationException("RabbitMQ message bus is not connected.");
        }
 
        lock (_syncRoot)
        {
            if (!_handlers.TryGetValue(route, out var handlers))
            {
                handlers = [];
                _handlers[route] = handlers;
            }
 
            handlers.Add(handler);
        }
 
        _channel.QueueBind(
            queue: _options.QueueName,
            exchange: _options.ExchangeName,
            routingKey: route);
 
        if (!string.IsNullOrWhiteSpace(_consumerTag))
        {
            return Task.CompletedTask;
        }
 
        _consumerTag = _channel.BasicConsume(
            queue: _options.QueueName,
            autoAck: false,
            onReceived: (delivery, token) => HandleReceivedAsync(
                delivery,
                token.CanBeCanceled ? token : cancellationToken));
 
        return Task.CompletedTask;
    }

다음을 참조함 : _channel, _consumerTag, _handlers, _options, _syncRoot, HandleReceivedAsync(), State.

이 함수 내부에서 호출하는 함수들에 대한 그래프입니다.:

ValidateOptions()

void Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.ValidateOptions ( RabbitMqMessageBusOptions options )

inlinestaticprivate

필수 RabbitMQ 연결 및 토폴로지 설정과 포트 범위를 검증합니다.

매개변수

options

검증할 메시지 버스 설정입니다.

예외

ArgumentException	필수 문자열 설정이 비어 있는 경우 발생합니다.
ArgumentOutOfRangeException	포트가 1~65535 범위를 벗어난 경우 발생합니다.

RabbitMqMessageBus.cs 파일의 701 번째 라인에서 정의되었습니다.

    {
        ArgumentException.ThrowIfNullOrWhiteSpace(options.HostName);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.VirtualHost);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.UserName);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.Password);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.ExchangeName);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.QueueName);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.RoutingKey);
        ArgumentException.ThrowIfNullOrWhiteSpace(options.ExchangeType);
 
        if (options.Port <= 0 || options.Port > 65535)
        {
            throw new ArgumentOutOfRangeException(nameof(options.Port));
        }
    }

다음을 참조함 : Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.ExchangeName, Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.ExchangeType, Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.HostName, Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.Password, Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.Port, Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.QueueName, Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.RoutingKey, Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.UserName, Dreamine.Communication.RabbitMQ.Options.RabbitMqMessageBusOptions.VirtualHost.

다음에 의해서 참조됨 : RabbitMqMessageBus().

이 함수를 호출하는 함수들에 대한 그래프입니다.:

멤버 데이터 문서화

_channel

IRabbitMqChannel? Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus._channel

private

channel 값을 보관합니다.

RabbitMqMessageBus.cs 파일의 82 번째 라인에서 정의되었습니다.

다음에 의해서 참조됨 : Cleanup(), ConnectAsync(), HandleReceivedAsync(), PublishAsync(), SubscribeAsync().

_connection

IRabbitMqConnection? Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus._connection

private

connection 값을 보관합니다.

RabbitMqMessageBus.cs 파일의 73 번째 라인에서 정의되었습니다.

다음에 의해서 참조됨 : Cleanup(), ConnectAsync().

_connectionFactory

readonly IRabbitMqConnectionFactory Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus._connectionFactory

private

connection Factory 값을 보관합니다.

RabbitMqMessageBus.cs 파일의 45 번째 라인에서 정의되었습니다.

다음에 의해서 참조됨 : ConnectAsync(), RabbitMqMessageBus().

_consumerTag

string? Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus._consumerTag

private

consumer Tag 값을 보관합니다.

RabbitMqMessageBus.cs 파일의 91 번째 라인에서 정의되었습니다.

다음에 의해서 참조됨 : Cleanup(), SubscribeAsync().

_handlers

readonly Dictionary<string, List<Func<MessageEnvelope, CancellationToken, Task> > > Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus._handlers = new()

private

handlers 값을 보관합니다.

RabbitMqMessageBus.cs 파일의 54 번째 라인에서 정의되었습니다.

다음에 의해서 참조됨 : Cleanup(), HandleReceivedAsync(), SubscribeAsync().

_options

readonly RabbitMqMessageBusOptions Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus._options

private

options 값을 보관합니다.

RabbitMqMessageBus.cs 파일의 36 번째 라인에서 정의되었습니다.

다음에 의해서 참조됨 : ConnectAsync(), DeclareTopology(), HandleReceivedAsync(), PublishAsync(), RabbitMqMessageBus(), SubscribeAsync().

_syncRoot

readonly object Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus._syncRoot = new()

private

sync Root 값을 보관합니다.

RabbitMqMessageBus.cs 파일의 63 번째 라인에서 정의되었습니다.

다음에 의해서 참조됨 : Cleanup(), HandleReceivedAsync(), SubscribeAsync().

속성 문서화

Kind

TransportKind Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.Kind

get

RabbitMQ 전송 방식을 가져옵니다.

RabbitMqMessageBus.cs 파일의 173 번째 라인에서 정의되었습니다.

State

ConnectionState Dreamine.Communication.RabbitMQ.Buses.RabbitMqMessageBus.State = ConnectionState.Disconnected

getprivate set

메시지 버스의 현재 연결 상태를 가져옵니다.

RabbitMqMessageBus.cs 파일의 163 번째 라인에서 정의되었습니다.

{ get; private set; } = ConnectionState.Disconnected;

다음에 의해서 참조됨 : ConnectAsync(), DisconnectAsync(), PublishAsync(), SubscribeAsync().

---

이 클래스에 대한 문서화 페이지는 다음의 파일로부터 생성되었습니다.:

• RabbitMqMessageBus.cs