MongolianBarbecue/Consumer.cs
/// <summary>
/// Gets the next available message or immediately returns null if no message was available
/// </summary>
Gets the next available message or immediately returns null if no message was available
Gets the next available message or immediately returns null if no message was available
public async Task<ReceivedMessage> GetNextAsync()
        {
            var now = DateTime.UtcNow;
            var receiveTimeCriteria = new BsonDocument { { "$lt", now.Subtract(_config.DefaultMessageLease) } };
            var filter = new BsonDocumentFilterDefinition<BsonDocument>(new BsonDocument
            {
                {Fields.DestinationQueueName, QueueName},
                {Fields.ReceiveTime, receiveTimeCriteria},
                {Fields.DeliveryAttempts, new BsonDocument {{"$lt", _config.MaxDeliveryAttempts}}},
            });
            var update = new BsonDocumentUpdateDefinition<BsonDocument>(new BsonDocument
            {
                {"$set", new BsonDocument {{Fields.ReceiveTime, now}}},
                {"$inc", new BsonDocument {{Fields.DeliveryAttempts, 1}}}
            });
            var options = new FindOneAndUpdateOptions<BsonDocument>
            {
                ReturnDocument = ReturnDocument.After
            };
            var collection = _config.Collection;
            await _semaphore.WaitAsync();
            try
            {
                var document = await collection.FindOneAndUpdateAsync(filter, update, options);
                if (document == null) return null;
                return GetReceivedMessage(document);
            }
            finally
            {
                _semaphore.Release();
            }
        }
public async Task < ReceivedMessage > GetNextAsync ( ) { var now = DateTime . UtcNow ; var receiveTimeCriteria = new BsonDocument { { " " , now . Subtract ( _config . DefaultMessageLease ) } } ; var filter = new BsonDocumentFilterDefinition < BsonDocument > ( new BsonDocument { { Fields . DestinationQueueName , QueueName } , { Fields . ReceiveTime , receiveTimeCriteria } , { Fields . DeliveryAttempts , new BsonDocument { { " " , _config . MaxDeliveryAttempts } } } , } ) ; var update = new BsonDocumentUpdateDefinition < BsonDocument > ( new BsonDocument { { " " , new BsonDocument { { Fields . ReceiveTime , now } } } , { " " , new BsonDocument { { Fields . DeliveryAttempts , 1 } } } } ) ; var options = new FindOneAndUpdateOptions < BsonDocument > { ReturnDocument = ReturnDocument . After } ; var collection = _config . Collection ; await _semaphore . WaitAsync ( ) ; try { var document = await collection . FindOneAndUpdateAsync ( filter , update , options ) ; if ( document == null ) return null ; return GetReceivedMessage ( document ) ; } finally { _semaphore . Release ( ) ; } }
Gets the next available message or immediately returns null if no message was available
Gets the next available message or immediately returns null if no message was available