كيف تديرون حركة المرور الكثيفة مع Anypoint MQ؟

<!-- تدفق المنتج: HTTP → Anypoint MQ -->
<flow name="order-producer-flow">

  <http:listener
    path="/orders"
    allowedMethods="POST"
    config-ref="HTTP_Listener_config"/>

  <!-- التحقق من الرسالة -->
  <validation:is-not-null
    value="#[payload.orderId]"
    message="orderId zorunludur"/>

  <!-- النشر إلى MQ -->
  <anypoint-mq:publish
    config-ref="Anypoint_MQ_Config"
    destination="orders-exchange"
    messageId=#[payload.orderId]>
    <anypoint-mq:properties>
      <anypoint-mq:property
        key="priority"
        value=#[payload.priority default 'NORMAL']/>
      <anypoint-mq:property
        key="correlationId"
        value=#[correlationId]/>
    </anypoint-mq:properties>
  </anypoint-mq:publish>

  <!-- إرجاع 202 Accepted فورًا -->
  <set-payload
    value=#[output application/json --- {status: "queued", messageId: payload.orderId}]/>
  <http:response statusCode="202"/>

</flow>

<flow name="order-consumer-flow">

  <anypoint-mq:subscriber
    config-ref="Anypoint_MQ_Config"
    destination="orders-queue"
    maxConcurrency="10"
    ackMode="MANUAL"/>

  <!-- التكرارية: لا تعالج نفس الرسالة مرتين -->
  <os:retrieve
    config-ref="ObjectStore_Config"
    key=#[attributes.messageId]
    target="alreadyProcessed"
    defaultValue="false"/>

  <choice>
    <when expression=#[vars.alreadyProcessed == false]>
      <flow-ref name="process-order-subflow"/>
      <os:store
        config-ref="ObjectStore_Config"
        key=#[attributes.messageId]
        value="true"
        ttl="86400"
        ttlUnit="SECONDS"/>
      <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
    </when>
    <otherwise>
      <!-- رسالة مكررة: فقط ACK، لا معالجة -->
      <logger message="Duplicate mesaj atlandı: #[attributes.messageId]"/>
      <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
    </otherwise>
  </choice>

</flow>

<flow name="order-consumer-flow">

  <anypoint-mq:subscriber
    destination="orders-queue"
    ackMode="MANUAL"/>

  <try>
    <flow-ref name="process-order-subflow"/>
    <anypoint-mq:ack ackToken=#[attributes.ackToken]/>

    <error-handler>
      <!-- خطأ مؤقت: NACK، يعود إلى قائمة الانتظار -->
      <on-error-continue type="CONNECTIVITY, TIMEOUT">
        <logger message="Geçici hata, NACK: #[error.description]" level="WARN"/>
        <anypoint-mq:nack ackToken=#[attributes.ackToken]/>
      </on-error-continue>

      <!-- خطأ دائم: أرسل مباشرة إلى DLQ -->
      <on-error-continue type="VALIDATION, TRANSFORMATION">
        <anypoint-mq:publish destination="orders-dlq">
          <anypoint-mq:properties>
            <anypoint-mq:property key="errorReason" value=#[error.description]/>
            <anypoint-mq:property key="originalMessageId" value=#[attributes.messageId]/>
          </anypoint-mq:properties>
        </anypoint-mq:publish>
        <anypoint-mq:ack ackToken=#[attributes.ackToken]/>
      </on-error-continue>
    </error-handler>
  </try>

</flow>

// MQ'dan gelen sipariş → SAP formatına dönüşüm
%dw 2.0
output application/xml

var priorityMap = {
  "HIGH":   "01",
  "NORMAL": "02",
  "LOW":    "03"
}
---
SALESORDER: {
  HEADER: {
    ORDER_ID:    payload.orderId,
    CUSTOMER:   payload.customerId,
    PRIORITY:   priorityMap[attributes.properties."priority" default "NORMAL"],
    CREATED_AT: now() as String { format: "yyyyMMddHHmmss" },
    EXT_REF:    attributes.properties."correlationId" default ""
  },
  ITEMS: {
    (payload.items map (item, idx) -> {
      ITEM: {
        LINE_NO:  (idx + 1) * 10,
        MATERIAL: item.sku,
        QTY:      item.quantity,
        UNIT:     item.unit default "EA"
      }
    })
  }
}

<anypoint-mq:subscriber
  config-ref="Anypoint_MQ_Config"
  destination="orders-queue"
  <!-- عدد الرسائل التي تُعالَج بالتوازي -->
  maxConcurrency="20"
  <!-- عدد الرسائل المسحوبة في استطلاع واحد (1-10) -->
  fetchSize="10"
  <!-- المدة المنتظَرة إذا كانت قائمة الانتظار فارغة (ms) -->
  pollingTime="1000"
  <!-- مهلة معالجة الرسالة -->
  acknowledgementTimeout="60000"
  <!-- وضع ACK اليدوي -->
  ackMode="MANUAL"/>