Uma classe genérica de acesso ao MSMQ

O que é o MSMQ?

O MSMQ é o servidor nativo de troca de mensagens da Microsoft. Ele é disponibilizado juntamente com o CD de instalação do Windows, e sua instalação e configuração é extremamente fácil, tornando-se, pois, uma solução prática (e barata), quando se deseja fazer o processamento assíncrono de mensagens entre aplicações.

Uso do MSMQ

Um caso muito comum, é a integração de legados heterogêneos, onde podemos postar um texto (xml ou plain text) numa fila pré-definida, e o outro legado procede com a retirada da mensagem, e processamento. As filas MSMQ também podem conter objetos binários como mensagem.

Já desenvolvi diversas rotinas de processamento assíncrono com MSMQ, em diversas linguagens, mas esta última que criei é a que considero a mais simples e prática, principalmente pelo baixo acoplamento.

A classe de uso geral, para interagir com MSMQ

Analisando a situação:

a) Filas MSMQ podem ser transacionais, ou não.
b) Normalmente, uma rotina que envia mensagens ao MSMQ deve garantir sua transação completa, por exemplo, você deseja que sua mensagem seja considerada enviada apenas depois de ter feito diversos processamentos, após a postagem na fila. Caso contrário, seu processamento poderá enviar 2x a mesma mensagem.
c) Da mesma maneira, uma rotina que retira mensagens do MSMQ deve garantir que a mensagem retirada foi corretamente processada.

Enunciado da solução:

a) A classe de processamento MSMQ não deve conter lógica explicita de processamento da mensagem postada, ou seja, você deve "plugar" o processamento relacionado ao pós-envio da mensagem para o método executar. O mesmo vale para o método de retirada de mensagens.
Para este código, foi adotado o ActiveXMessageFormatter, que usa um formato compatível com o antigo MSMQ ActiveX Component. Outros formatters podem ser usados, aliás, um bom exercício para uma extensão da classe seria poder determinar o tipo de formatter a utilizar.
b) A inteligência de processamento deve ser passada para os métodos, através de um delegate. Assim, temos 2 delegates distintos:

public delegate bool MSMQProcessaMensagemRecebida(Message mensagemAtual);

Onde Message é um objeto do namespace System.Messaging, que representa uma mensagem MSMQ.

public delegate bool MSMQProcessaAposMensagemEnviada(String corpoMensagem, String labelMensagem);

Onde são passados como parâmetro os mesmos conteúdos para o corpo da mensagem e o label. O Label é um identificador da mensagem, o mesmo aparece na visualização de mensagens da fila, e é, na maioria das vezes, alimentado com algum código que possa ser associado ao conteúdo postado.

Observem que os 2 delegates retornam um bool. Se o retorno for true, indica-se que o processamento foi completado com sucesso.

Segue o código da classe, comentado:

using System; 
using System.Collections.Generic; 
using System.Linq; 
using System.Text; 
using System.Messaging; // namespace que contempla as classes MSMQ 

namespace RotinasGenericas 
{ 
public class MSMQ 
{ 

#region propriedades 

public string NomeFila 
{ 
get; 
private set; 
} 

public bool Transacional 
{ 
get; 
private set; 
} 

#endregion 

#region declaracao de delegates 
public delegate bool MSMQProcessaMensagemRecebida(Message mensagemAtual); 
public delegate bool MSMQProcessaAposMensagemEnviada(String corpoMensagem, String labelMensagem); 
#endregion 

public MSMQ(String nomeFila, bool bTransacional) 
{ 
// passamos no construtor a característica de fazer um 
// processamento transacional da mensagem enviada. 
this.NomeFila = nomeFila; 
this.Transacional = bTransacional; 
} 

// Processa, com base no enumerador de mensagens atual, 
// a retirada da mesma, dentro de uma transação ou não. 
// Retorna a mensagem da posição atual do enumerador. 
private Message obtemMensagem(MessageEnumerator msmqEnumerator, 
MessageQueueTransaction msmqTransacao) 
{ 
Message msmqRetorno = msmqEnumerator.Current; 
msmqRetorno.Formatter = new ActiveXMessageFormatter(); 
if (msmqTransacao != null) 
msmqEnumerator.RemoveCurrent(msmqTransacao); 
else 
msmqEnumerator.RemoveCurrent(); 
return msmqRetorno; 
} 

/// 
/// Processa o recebimento de todas as mensagens represadas na fila do MSMQ 
/// 
/// Delegate. Metodo que vai ser chamado, a cada mensagem lida da fila. 
public void receberMensagens(MSMQProcessaMensagemRecebida metodoProcessar) 
{ 
MessageQueue msmqFila = null; 
MessageQueueTransaction msmqTransacao = null; 
MessageEnumerator msmqEnumerator = null; 
Message msmqMensagemCorrente = null; 
bool retornoProcessamento = false; 
try 
{ 
msmqMensagemCorrente = null; 
msmqFila = new MessageQueue(this.NomeFila); // abre a fila 
// obtem o rol de mensagens atualmente postadas na fila.
msmqEnumerator = msmqFila.GetMessageEnumerator2();  
retornoProcessamento = false; // variável de retorno do processamento 

while (msmqEnumerator.MoveNext()) // enquanto existirem mensagens na fila 
{ 
// abre-se a transação, caso a instância tenha sido criada como TRANSACIONAL 
// Reparem, que temos uma transação para cada mensagem que iremos processar. 
if (this.Transacional) 
{ 
msmqTransacao = new MessageQueueTransaction(); 
msmqTransacao.Begin(); // Inicializa-se a transação. 
} 

// Obtemos a mensagem corrente da fila. 
msmqMensagemCorrente = this.obtemMensagem(msmqEnumerator, msmqTransacao); 

// CASO tenhamos passado uma instância do delegate de processamento, 
// executamos a chamada do mesmo. Caso contrário, assumimos que o 
// processamento foi terminado com sucesso. 
retornoProcessamento = (metodoProcessar != null) ? metodoProcessar(msmqMensagemCorrente) : true; 

// se estamos numa transação 
if (this.Transacional) 
{ 

// caso tenhamos processado com sucesso, fazemos o commit, e a mensagem é retirada da fila. 
// caso contrário, é feito o Abort, e a mensagem permanece na fila. 
if (retornoProcessamento) msmqTransacao.Commit(); 
else msmqTransacao.Abort(); 
msmqTransacao.Dispose(); 
msmqTransacao = null; 
} 

msmqMensagemCorrente.Dispose(); 
msmqMensagemCorrente = null; 
} 
} 
catch (Exception erro) 
{ 
// em caso de erro, é feito o abort da transação, e a mensagem permanece na fila. 
if (msmqTransacao != null) msmqTransacao.Abort(); 
throw erro; 
} 
finally 
{ 
// liberação dos recursos. 
if (msmqMensagemCorrente!=null) msmqMensagemCorrente.Dispose(); 
if (msmqTransacao != null) msmqTransacao.Dispose(); 
if (msmqEnumerator != null) msmqEnumerator.Dispose(); 
if (msmqFila != null) msmqFila.Dispose(); 
msmqFila = null; 
msmqEnumerator = null; 
msmqMensagemCorrente = null; 
msmqTransacao = null; 
} 
} 


/// 
/// Processa o envio da mensagem enviada como parƒmetro. 
/// 
/// Corpo da mensagem a enviar 
/// Identificador da Mensagem 
/// Metodo a processar, a cada mensagem enviada para a fila. 
/// 
public bool enviarMensagem(String corpoMensagem, 
String labelMensagem,MSMQProcessaAposMensagemEnviada metodoProcessar) 
{ 
bool boolRet = false; 
MessageQueue msmqFila = null; 
MessageQueueTransaction msmqTransacao = null; 
try 
{ 
msmqFila = new MessageQueue(this.NomeFila); // abre a fila. 
msmqFila.Formatter = new ActiveXMessageFormatter(); 

if (this.Transacional) 
{ 
// abre a transação 
msmqTransacao = new MessageQueueTransaction(); 
msmqTransacao.Begin(); 
msmqFila.Send(corpoMensagem, labelMensagem, msmqTransacao); 
} 
else // envia sem transação. 
msmqFila.Send(corpoMensagem, labelMensagem); 

// caso tenhamos passado o delegate, processamos o mesmo. Caso contrário, assumimos true. 
boolRet = (metodoProcessar != null) ? metodoProcessar(corpoMensagem,labelMensagem) : true; 

// caso estejamos em transação, processamos o Commit ou Abort, 
// dependendo do resultado do processamento. 
if (this.Transacional) 
{ 
if (boolRet) msmqTransacao.Commit(); 
else msmqTransacao.Abort(); 
} 
} 
catch (Exception err) 
{ 
// caso tenha ocorrido um erro de processamento, e a mensagem tenha sido 
// postada na fila, é feito o rollback da transação. 
if (msmqTransacao != null) 
{ 
if (msmqTransacao.Status == MessageQueueTransactionStatus.Pending) 
msmqTransacao.Abort(); 
} 
throw err; 
} 
finally 
{ 
if (msmqTransacao != null) msmqTransacao.Dispose(); 
if (msmqFila != null) msmqFila.Dispose(); 
msmqFila = null; 
msmqTransacao = null; 
} 
return boolRet; 
} 

} 
}

Vejam que o código é muito simples, e os delegates garantem que o mesmo seja reutilizável por várias classes que precisem interagir com o MSMQ.

Abaixo, um exemplo da utilização da classe:
Temos um form, onde temos 2 caixas de texto, com os nomes das filas de envio e recebimento (por exemplo, .\private$\teste_envio e .\private$\teste_retorno), 2 caixas de texto, com o conteúdo da mensagem a enviar ou recebida, e 2 checkboxes, que indicam se o envio é transacional, ou não.

using System; 
using System.Collections.Generic; 
using System.ComponentModel; 
using System.Data; 
using System.Drawing; 
using System.Linq; 
using System.Text; 
using System.Windows.Forms; 

namespace TesteEnvioMSMQ 
{ 
public partial class Form1 : Form 
{ 
public Form1() 
{ 
InitializeComponent(); 
} 
private bool AposEnvio(String corpoMensagem, String labelMensagem) 
{ 
MessageBox.Show("Mensagem enviada com sucesso."); 
return true; 
} 

private bool AposRetorno(System.Messaging.Message mensagem) 
{ 
this.txtConteudoLido.Text = mensagem.Body.ToString(); 
return true; 
} 
private void btnEnviar_Click(object sender, EventArgs e) 
{ 
CRK.FRT.BP.Generics.MSMQ objFila = new RotinasGenericas.MSMQ(this.txtFilaENVIO.Text, 
this.chkEnvioTransacional.Checked); 
objFila.enviarMensagem(this.txtConteudoEnviar.Text, Guid.NewGuid().ToString(), new 
RotinasGenericas.MSMQ.MSMQProcessaAposMensagemEnviada (this.AposEnvio)); 
} 
private void btnLer_Click(object sender, EventArgs e) 
{ 
CRK.FRT.BP.Generics.MSMQ objFila = new RotinasGenericas.MSMQ(this.txtFilaRETORNO.Text,
 this.chkRetornoTransacional.Checked); 
objFila.receberMensagens(new RotinasGenericas.MSMQ.MSMQProcessaMensagemRecebida(this.AposRetorno)); 
} 
} 
}

A codificação dos delegates ficam a critério do aplicativo cliente, o processamento do MSMQ fica encapsulado, e é completamente independente desta lógica.

Espero que seja de boa valia, para vocês. Mais dicas e códigos, no meu blog, http://taotecnologia.blogspot.com.

Desenvolvimento - C#

Leia também