智慧代理

脈絡：一對 Wire Tap 可以用來追蹤流經某個元件的訊息，但這個作法假設該元件把訊息發布到一個固定的輸出通道。然而許多服務式元件是把回覆訊息發布到「請求訊息中所指定的 Return Address」。

難處在回覆這一邊#

要追蹤流經服務的訊息，我們得同時捕捉請求與回覆訊息。

用 Wire Tap 攔截請求訊息很容易
攔截回覆訊息才是難的部分——因為服務會依請求方偏好的 Return Address 把回覆發布到不同通道

多數 Request-Reply 服務都必須支援 Return Address，請求方才能指定回覆該送到哪個通道。若把服務改成永遠貼到固定通道，每個請求方就很難取出屬於自己的回覆訊息。
有些訊息系統允許消費者在單一回覆佇列中「peek」特定訊息，但那是實作專屬的作法，而且在「回覆不是回給請求方、而是給第三方」的情況下行不通。

為什麼不改元件本身#

如同 Wire Tap 中討論過的，修改元件來檢視訊息未必可行或實際：面對自製應用時我們可能改不了程式碼，得實作一個位於應用程式之外的方案。
而且我們也未必想要求每個應用程式都實作檢視邏輯——尤其這些邏輯的性質可能隨「測試模式」或「正式模式」而異。
把檢視功能保持在一個獨立、自我完備的元件中，能提升彈性、重用性與可測試性。

解法#

流程是：

Smart Proxy 攔截送往 Request-Reply 服務之請求通道上的訊息
對每則進站訊息，保存原始寄件端所指定的 Return Address
把訊息中的 Return Address 換成 Smart Proxy 自己所監聽的回覆通道
回覆訊息抵達該通道時，取出先前保存的 Return Address，並用 Message Router 把未修改的回覆轉送到那個通道

保存資料的兩種位置#

Smart Proxy 必須以「能把進站回覆訊息與 Return Address 關聯起來」的方式保存資料。它可以存在兩個地方：

存在訊息裡#

Smart Proxy 在訊息中新增一個帶有 Return Address 的欄位，並要求 Request-Reply 服務把這個欄位複製到回覆訊息中。之後它只需從回覆中取出這個特殊欄位、把欄位移除，再把訊息轉送到該欄位所指定的通道。

存在 Smart Proxy 裡#

把 Return Address 存進專用儲存（記憶體結構或關聯式資料庫）。

因為 Smart Proxy 的目的本來就是追蹤請求與回覆之間的訊息，它通常反正都得保存請求訊息的資料，才能把它與回覆關聯起來並一併分析。

這個作法要求 Smart Proxy 能把回覆訊息與請求訊息關聯起來。多數 Request-Reply 服務支援 Correlation Identifier（服務把它從請求複製到回覆）；若 Smart Proxy 無法修改原始訊息格式，它可以（濫）用這個欄位來做關聯。

不是所有請求方都會指定 Correlation Identifier
Correlation Identifier 原本只需在「單一請求方所發的各個請求之間」唯一，而不需要跨多個請求方唯一——而現在服務的回覆佇列上承載著來自多個請求方的訊息，沿用原始的 Correlation Identifier 就不可靠了

因此 Smart Proxy 把原始的 Correlation Identifier 與原始 Return Address 一起保存，並用自己的 Correlation Identifier 取代原始的，這樣回覆訊息抵達時才能取回原始的兩者。

一個額外的麻煩#

因此 Smart Proxy 必須把回覆訊息中的這個 Correlation Identifier，換成「原始請求訊息的 Message ID」，請求方才能正確地把請求與回覆關聯起來。

圖 11-9：保存並替換 Correlation Identifier 與 Return Address

範例：用 MSMQ 與 C# 實作簡單的 Smart Proxy

實作 Smart Proxy 並不像聽起來那麼複雜。以下場景由兩個請求方、一個 Smart Proxy 與一個簡單服務組成，Smart Proxy 把訊息處理時間送到控制匯流排供主控台顯示。

共用基底：MessageConsumer#

為了寫程式方便，先定義一個封裝「建立事件驅動訊息消費者」所需程式碼的基底類別。繼承的類別只要覆寫虛擬方法 ProcessMessage，完全不必操心佇列設定與事件驅動處理——把這段程式碼抽成共同基底類別，讓「用幾行程式碼做出測試客戶端與假的請求－回覆服務」變得很容易。

public class MessageConsumer
{
    protected MessageQueue inputQueue;

    public MessageConsumer (MessageQueue inputQueue)
    {
        this.inputQueue = inputQueue;
        SetupQueue(this.inputQueue);
        Console.WriteLine(this.GetType().Name + ": Processing messages from " + inputQueue.Path);
    }

    protected void SetupQueue(MessageQueue queue)
    {
        queue.Formatter = new System.Messaging.XmlMessageFormatter(
            new String[] {"System.String,mscorlib"});
        queue.MessageReadPropertyFilter.ClearAll();
        queue.MessageReadPropertyFilter.AppSpecific = true;
        queue.MessageReadPropertyFilter.Body = true;
        queue.MessageReadPropertyFilter.CorrelationId = true;
        queue.MessageReadPropertyFilter.Id = true;
        queue.MessageReadPropertyFilter.ResponseQueue = true;
    }

    public virtual void Process()
    {
        inputQueue.ReceiveCompleted += new ReceiveCompletedEventHandler(OnReceiveCompleted);
        inputQueue.BeginReceive();
    }

    private void OnReceiveCompleted(Object source, ReceiveCompletedEventArgs asyncResult)
    {
        MessageQueue mq = (MessageQueue)source;

        Message m = mq.EndReceive(asyncResult.AsyncResult);
        m.Formatter = new System.Messaging.XmlMessageFormatter(
            new String[] {"System.String,mscorlib"});

        ProcessMessage(m);

        mq.BeginReceive();
    }

    protected virtual void ProcessMessage(Message m)
    {
        String text = "";
        try
        {
            text = (String)m.Body;
        }
        catch (InvalidOperationException) {};
        Console.WriteLine(this.GetType().Name + ": Received Message " + text);
    }
}

SmartProxy#

Smart Proxy 含有兩個 MessageConsumer——一個處理來自請求方的請求訊息、一個處理服務回傳的回覆訊息——外加一個在請求與回覆之間保存訊息資料的 Hashtable：

public class SmartProxyBase
{
    protected SmartProxyRequestConsumer requestConsumer;
    protected SmartProxyReplyConsumer replyConsumer;

    protected Hashtable messageData;

    public SmartProxyBase(MessageQueue inputQueue, MessageQueue serviceRequestQueue,
                          MessageQueue serviceReplyQueue)
    {
        messageData = Hashtable.Synchronized(new Hashtable());
        requestConsumer = new SmartProxyRequestConsumer(inputQueue, serviceRequestQueue,
                                                       serviceReplyQueue, messageData);
        replyConsumer = new SmartProxyReplyConsumer(serviceReplyQueue, messageData);
    }

    public virtual void Process()
    {
        requestConsumer.Process();
        replyConsumer.Process();
    }
}

SmartProxyRequestConsumer#

它相對簡單：把請求訊息的相關資訊（message ID、Return Address、AppSpecific 屬性與當前時間）存進 hashtable，索引鍵是「送給實際服務之新請求訊息的 message ID」。

Request-Reply 服務會把這個 message ID 複製到服務回覆訊息的 CorrelationID 欄位，Smart Proxy 因而能取回所存的訊息資料。

它同時把 Return Address（ResponseQueue 屬性）換成 Smart Proxy 所監聽的回覆佇列；另外留了一個虛擬方法 AnalyzeMessage 供子類別做任何想要的分析：

public class SmartProxyRequestConsumer : MessageConsumer
{
    protected Hashtable messageData;
    protected MessageQueue serviceRequestQueue;
    protected MessageQueue serviceReplyQueue;

    public SmartProxyRequestConsumer(MessageQueue requestQueue,
        MessageQueue serviceRequestQueue, MessageQueue serviceReplyQueue,
        Hashtable messageData) : base(requestQueue)
    {
        this.messageData = messageData;
        this.serviceRequestQueue = serviceRequestQueue;
        this.serviceReplyQueue = serviceReplyQueue;
    }

    protected override void ProcessMessage(Message requestMsg)
    {
        base.ProcessMessage(requestMsg);

        MessageData data = new MessageData(requestMsg.Id, requestMsg.ResponseQueue,
                                           requestMsg.AppSpecific);
        requestMsg.ResponseQueue = serviceReplyQueue;
        serviceRequestQueue.Send(requestMsg);
        messageData.Add(requestMsg.Id, data);
        AnalyzeMessage(requestMsg);
    }

    protected virtual void AnalyzeMessage(Message requestMsg)
    {
    }
}

SmartProxyReplyConsumer#

它監聽服務的回覆通道：ProcessMessage 取回由請求端保存的訊息資料、呼叫 AnalyzeMessage 樣板方法，接著把 CorrelationID 與 AppSpecific 屬性複製到新的回覆訊息，並把它路由到原始請求訊息所指定的 Return Address。