Correlación de los mensajes de agregación de la petición de entrada y de la respuesta de salida

Si está utilizando WebSphere MQ como el protocolo de transporte para los flujos de agregación, puede correlacionar el mensaje de petición inicial con el mensaje de respuesta combinado conservando el MsgId del MQMD de entrada y utilizándolo como el CorrelId de la respuesta de salida.

Antes de comenzar:

Para realizar esta tarea, debe haber completado las tareas siguientes:

Cuando el flujo ramificado procesa la petición inicial, se pierde el MsgId que se conserva en MQMD. Si desea correlacionar el mensaje de salida agregado con el mensaje de entrada original, un modo de hacerlo es conservar el MsgId y especificarlo como el CorrelId en el mensaje de salida agregado.

En la tarea descrita se presupone que ya ha creado un flujo ramificado y un flujo de abanico de entrada como dos flujos distintos que se ejecutan en el mismo intermediario. También presupone el contenido mínimo para estos flujos. Los flujos pueden incluir el contenido que resulte adecuado y puede adaptarlos del modo que se muestra aquí para añadir esta función adicional.

  1. Vaya a la perspectiva Desarrollo de aplicaciones de intermediario.
  2. Abra el flujo ramificado y añada una nueva ramificación que incluya estos nodos:
    • Un nodo Compute llamado SaveMsgId, conectado al terminal de salida del nodo AggregateControl.
    • Un nodo MQOutput llamado SavedMsgIdQ, conectado al terminal de salida de SaveMsgId
    • Un nodo AggregateRequest, conectado al terminal de salida de SavedMsgIdQ
    A continuación se muestra el flujo ramificado. Ahora tiene tres bifurcaciones del nodo AggregateControl:
    1. La nueva ramificación con el nodo Compute, llamado SaveMsgId, que añade un MQMD a una copia del mensaje de entrada y conserva el MsgId.
    2. La ramificación original que procesa la petición. Esto se muestra sólo con unHandleRequest de nodo Compute que extrae una sola petición de la entrada combinada, añade un MQMD y emite la nueva petición a RequestQ del nodo MQOutput. Este va seguido de un nodo AggregateRequest que registra que esta petición se ha emitido.
    3. La ramificación de control, en la que un Add_MQMD de un nodo Compute añade un MQMD al mensaje de control y lo pasa a ControlQ de un nodo MQOutput para entregarlo al nodo AggregateReply de un flujo de abanico de entrada diferente.

    Muestra un flujo ramificado mínimo (el nodo MQInput conectado al nodo AggregateControl, conectado al flujo de control (el nodo Compute que añade un MQMD conectado a MQOutput), el flujo normal (el nodo Compute conectado a MQOutput, conectado a AggregateRequest). También tiene un flujo adicional para conservar el MsgId (el nodo Compute conectado a MQOutput conectado a AggregateRequest).
  3. Codificar el ESQL del nuevo SaveMsgId del nodo Compute, como se muestra a continuación:
    CREATE COMPUTE MODULE SaveMsgId
      CREATE FUNCTION Main() RETURNS BOOLEAN
      BEGIN
        CALL CopyMessageHeaders();
    
    // Extraer y conservar el MsgId del mensaje de petición original
        SET OutputRoot.XML.MyMsgId = InputRoot.MQMD.MsgId;
      END;
      CREATE PROCEDURE CopyMessageHeaders()
      BEGIN
        DECLARE I INTEGER 1;
        DECLARE J INTEGER CARDINALITY(InputRoot.*[]);
        WHILE I < J DO
          	SET OutputRoot.*[I] = InputRoot.*[I];
          SET I  = I + 1;
        END WHILE;
      END;
    END MODULE;
  4. Configurar SavedMsgIdQ del nodo MQOutput:
    • Establezca la propiedad Nombre de cola en SAVEDMSGID.
    • Seleccione Petición en el navegador del diálogo de propiedades y seleccione el recuadro de selección Petición.
    • Escriba el nombre REPLIES para la propiedad Cola de respuestas. Este es el nombre de la cola a la que se envían los mensajes de respuesta como respuesta a las peticiones enviadas por el flujo ramificado.
  5. Configure el nodo AggregateRequest en la nueva ramificación estableciendo la propiedad Carpeta en SavedMSGID
  6. Crear un flujo de mensajes nuevo que contenga:
    • Un SavedMsgId del nodo MQInput, con la propiedad Nombre de cola establecida en SAVEDMSGID
    • Un nodo MQReply, conectado al terminal de salida de SavedMsgId
    Esto se muestra a continuación:
    Aquí se muestra el flujo nuevo que recibe el mensaje que contiene el MsgId de MQMD y lo direcciona a la cola de entrada REPLIES del flujo de abanico de entrada mediante un nodo MQReply
  7. Abrir el flujo del abanico de entrada y actualizar su nodo Compute para que maneje la nueva función. Añada el ESQL que se muestra a continuación en la función principal:
    SET OutputRoot.MQMD.Version = MQMD_CURRENT_VERSION; 
    // Almacenar el valor MsgId de MQMD de la carpeta especificada en el campo MQMD CorrelId
        SET OutputRoot.MQMD.CorrelId =
            CAST(InputRoot.ComIbmAggregateReplyBody.SavedMSGID.XML.MyMsgId AS BLOB);  
    A continuación, se muestra el flujo ramificado. El SetCorrelID del nodo Compute procesa las respuestas para crear un mensaje de respuesta combinado y ahora establece el identificador de correlación MQMD para asociar esta respuesta con la petición inicial:
    Muestra un flujo de abanico de entrada mínimo (el nodo MQInput que recibe el mensaje de control conectado al terminal de control del nodo AggregateReply, el nodo MQInput que recibe las respuestas conectado a AggregateReply en el terminal. El terminal de salida del nodo AggregateReply conectado a SetCorrelId del nodo Compute conectado a MQOutput que proporciona la respuesta combinada).
Esta tarea le muestra un ejemplo de cómo se conserva el MsgId de MQMD para correlacionar la petición inicial con la respuesta final. Se ha añadido una función adicional en la nueva ramificación de un flujo ramificado existente, de este modo, se añade otro mensaje a la petición y otro flujo de mensajes en el intermediario. Puede obtener el mismo resultado si añade el código para conservar el MsgId en una ramificación existente del flujo ramificado.

Conceptos relacionados
Flujos de mensajes
Agregación de flujos de mensajes

Tareas relacionadas
Configuración de flujos de agregación
Diseño de un flujo de mensajes
Creación de un flujo de mensajes
Definición del contenido del flujo de mensajes

Referencia relacionada
Nodo AggregateControl
Nodo AggregateReply
Nodo AggregateRequest