wcf msmq listener service удаляет сообщения, но не обрабатывает

Question

У меня есть служба WCF, которая создает личное сообщение в MSMQ. Сообщения создаются отлично, и я вижу сообщения, которые создает мое клиентское приложение.

У меня также есть служба прослушивания MSMQ, которая, кажется, работает нормально. Когда я запускаю службу, она, похоже, успешно «обрабатывает» сообщения в очереди, и они удаляются. Однако реализация моей службы прослушивателя, похоже, не выполняется.

Я довольно новичок в MSMQ, и я понятия не имею, почему сообщения удаляются из очереди и почему код в моем методе прослушивания не выполняется.

Ниже приведен мои классы услуг ...

[ServiceContract(Namespace = ServiceConstants.NAMESPACE, Name = "IOrderService")] 
public interface IOrderQueueProcessingService 
{ 

    [OperationContract(IsOneWay = true, Action = "*")] 
    void ProcessOrderQueue(MsmqMessage<string> message); 

} 

public abstract class OrderQueueProcessingServiceBase : ServiceBase, IOrderQueueProcessingService 
{ 
    #region CONSTRUCTORS 

    protected OrderQueueProcessingServiceBase() { } 

    protected OrderQueueProcessingServiceBase(List<EventRecord> existingList) : base(existingList) { } 

    #endregion //CONSTRUCTORS 

    #region IOrderQueueProcessingService Members 


    [OperationBehavior(TransactionScopeRequired = false, TransactionAutoComplete = true)] 
    public virtual void ProcessOrderQueue(MsmqMessage<string> message) 
    { 
     throw new NotImplementedException(); 
    } 

    #endregion 
} 

public class OrderQueueProcessingService : OrderQueueProcessingServiceBase 
{ 
    #region Constructors 


    public OrderQueueProcessingService() {} 

    public OrderQueueProcessingService(List<EventRecord> existingList) : base(existingList) { } 


    #endregion 

    /// <summary> 
    ///  Processes any Orders in the Orders Queue 
    /// </summary> 
    /// <param name="message"></param> 
    public override void ProcessOrderQueue(MsmqMessage<string> message) 
    { 

     var q = new MessageQueue(@".\private$\msmqdemo/submitorderservice.svc"); 
     q.Send("hey"); 
     /* 
     using (new Tracer("OrderQueueProcessingService")) 
     { 
      // add data context to work with. 
      using (var unitOfWork = new TLFDataContext()) 
      { 

       var newOrderLines = new List<OrderLineDataContract> 
           { 
            new OrderLineDataContract 
             { 
              C = "test", 
              IC = "msw", 
              Qty = 1, 
              T = "volume" , 
              ED = DateTime.UtcNow.AddDays(5) 
             } 
           }; 

       var newOrder = new OrderDataContract 
       { 
        LIs = newOrderLines.AsEnumerable(), 
        PId = 9323, 
        POId = 8686, 
        S = "new" 

       }; 
       var orderService = new OrderService(); 
       var createdOrder = orderService.CreateOrder(null, null, newOrder); 
       //unitOfWork.SubmitUnitOfWork(); 

       //return null; 
      } 
     }*/ 


    } 

} 

Я закомментирован код, который я в конце концов пытаюсь выполнить, и заменить его с простым MSMQ отправить сообщение, для тестирования. Кажется, это должно работать нормально. Любая помощь будет принята с благодарностью.

настройки Config ниже ...

<service name="ServiceImplementation.OrderQueueProcessingService"> 
    <host> 
     <baseAddresses> 
     <add baseAddress="https://localhost:9000/OrderQueueProcessingService.svc" /> 
     </baseAddresses> 
    </host> 
    <endpoint address="net.msmq://localhost/private/testingqueue/OrderQueueProcessingService.svc" binding="netMsmqBinding" bindingConfiguration="MsmqBindingNonTransactionalNoSecurity" contract="IOrderQueueProcessingService" /> 
    </service> 

Answer 1

Я был в состоянии понять мой вопрос. У меня не было обработки событий для события QueueOnPeekComplted. Я добавил это к методу Initialize и смог успешно обработать мои сообщения. Я также добавил обработку сообщений, которые не могли быть обработаны. Ниже приведена моя новая реализация моего сервиса OrderQueueProcessingService.

public class OrderQueueProcessingService : OrderQueueProcessingServiceBase, IDisposable 
{ 
    #region Constructors 

    public OrderQueueProcessingService() 
    { 
     Initialize(ConfigurationManager.AppSettings["OrderQueueProcessingQueueName"]); 
    } 

    public OrderQueueProcessingService(List<EventRecord> existingList) : base(existingList) {} 

    #endregion 

    #region Properties 

    private MessageQueue Queue { get; set; } 

    #endregion 

    #region IDisposable Members 

    public new void Dispose() 
    { 
     if (Queue == null) return; 

     //unsubscribe and dispose 
     Queue.PeekCompleted -= QueueOnPeekCompleted; 
     Queue.Dispose(); 
    } 

    #endregion 

    private void Initialize(string queueName) 
    { 
     Queue = new MessageQueue(queueName, false, true, QueueAccessMode.Receive) 
        { 
          Formatter = new XmlMessageFormatter(new[] {typeof (OrderQueueDataContract)}) 
        }; 

     //setup events and start. 
     Queue.PeekCompleted += QueueOnPeekCompleted; 
     Queue.BeginPeek(); 
    } 

    private static void MoveMessageToDeadLetter(IDisposable message) 
    { 
     var q = new MessageQueue(ConfigurationManager.AppSettings["OrderProcessingDLQ"], QueueAccessMode.Send) 
        { 
          Formatter = new XmlMessageFormatter(new[] {typeof (OrderQueueDataContract)}) 
        }; 
     q.Send(message, MessageQueueTransactionType.Single); 
     q.Dispose(); 
    } 

    /// <summary> 
    ///  Processes the specified Order message 
    /// </summary> 
    /// <param name="orderMessage"></param> 
    public override void ProcessOrderQueue(OrderQueueDataContract orderMessage) 
    { 
     using (var unitOfWork = new MyDataContext()) 
     { 
      switch (orderMessage.M.ToLower()) 
      { 
       case "create": 
        DataAccessLayer.CreateOrder(unitOfWork, orderMessage.O.TranslateToBe()); 
        break; 
       default: 
        break; 
      } 
     } 
    } 

    private void QueueOnPeekCompleted(object sender, PeekCompletedEventArgs peekCompletedEventArgs) 
    { 
     var asyncQueue = (MessageQueue) sender; 

     using (var transaction = new MessageQueueTransaction()) 
     { 
      transaction.Begin(); 
      try 
      { 
       using (var message = asyncQueue.ReceiveById(peekCompletedEventArgs.Message.Id, TimeSpan.FromSeconds(30), transaction)) 
       { 
        if (message != null) ProcessOrderQueue((OrderQueueDataContract) message.Body); 
       } 
      } 
      catch (InvalidOperationException ex) 
      { 
       transaction.Abort(); 
      } 
      catch (Exception ex) 
      { 
       transaction.Abort(); 
      } 

      if (transaction.Status != MessageQueueTransactionStatus.Aborted) transaction.Commit(); 
      else 
      { 
       using (var message = asyncQueue.ReceiveById(peekCompletedEventArgs.Message.Id, TimeSpan.FromSeconds(30), transaction)) 
       { 
        if (message != null) 
        { 
         MoveMessageToDeadLetter(message); 
         message.Dispose(); 
        } 
       } 

       EventLog.WriteEntry("OrderQueueProcessingService", "Could not process message: " + peekCompletedEventArgs.Message.Id); 
      } 
      transaction.Dispose(); 
     } 

     asyncQueue.EndPeek(peekCompletedEventArgs.AsyncResult); 
     asyncQueue.BeginPeek(); 
    } 
} 

У кого-нибудь есть проблемы с этой реализацией? Мне пришлось немного поиграть с ним, но он прошел модульное тестирование и тестирование процессов.

Единственное, что я заметил, это то, что при первом запуске службы он обрабатывает все сообщения в очереди; что потенциально может задержать запуск моих других сервисов. Это может быть проблемой, когда я запускаю их в своем консольном приложении во время тестирования. Чтобы быть в безопасности, я запускаю этот сервис последним.