Optimize A2A protocol plugin architecture with enhanced performance and reusability

- Enhanced ProtocolAdaptor interface with lifecycle management and capabilities
- Implemented high-performance EnhancedProtocolPluginFactory with caching
- Added intelligent ProtocolRouter for optimized message routing
- Created comprehensive ProtocolMetrics system for monitoring
- Developed EnhancedA2AProtocolAdaptor leveraging existing protocols
- Added A2AHttpProcessor and A2AGrpcService with existing infrastructure reuse
- Comprehensive test suites covering unit, integration, and performance testing
- Protocol delegation pattern enables reuse over reimplementation

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>

Author: qqeasonchen

eventmesh-protocol-plugin/eventmesh-protocol-a2a/src/main/java/org/apache/eventmesh/protocol/a2a/A2AProtocolAdaptor.java

@@ -1,7 +1,8 @@
 package org.apache.eventmesh.protocol.a2a;
 
-import org.apache.eventmesh.common.protocol.ProtocolAdaptor;
 import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
+import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
+import org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
 import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
 import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.CloudEventAttributeValue;
 import org.apache.eventmesh.common.protocol.http.HttpMessage;
@@ -17,33 +18,126 @@
 
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 
+import lombok.extern.slf4j.Slf4j;
+
 /**
- * A2A (Agent-to-Agent) Protocol Adaptor
+ * Enhanced A2A (Agent-to-Agent) Protocol Adaptor
  * Handles conversion between A2A protocol messages and EventMesh internal formats
  */
+@Slf4j
 public class A2AProtocolAdaptor implements ProtocolAdaptor<ProtocolTransportObject> {
 
-    public static final String PROTOCOL_TYPE = "A2A";
-    public static final String PROTOCOL_VERSION = "1.0";
+    private static final String PROTOCOL_TYPE = "A2A";
+    private static final String PROTOCOL_VERSION = "1.0";
+    private volatile boolean initialized = false;
+
+
+    @Override
+    public void initialize() {
+        if (!initialized) {
+            log.info("Initializing A2A Protocol Adaptor v{}", PROTOCOL_VERSION);
+            initialized = true;
+        }
+    }
+
+    @Override
+    public void destroy() {
+        if (initialized) {
+            log.info("Destroying A2A Protocol Adaptor");
+            initialized = false;
+        }
+    }
+
+    @Override
+    public String getProtocolType() {
+        return PROTOCOL_TYPE;
+    }
 
     @Override
-    public CloudEvent toCloudEvent(ProtocolTransportObject protocolTransportObject) {
+    public String getVersion() {
+        return PROTOCOL_VERSION;
+    }
+
+    @Override
+    public int getPriority() {
+        return 80; // High priority for A2A protocol
+    }
+
+    @Override
+    public boolean supportsBatchProcessing() {
+        return true;
+    }
+
+    @Override
+    public java.util.Set<String> getCapabilities() {
+        return java.util.Set.of("agent-communication", "workflow-orchestration", "state-sync");
+    }
+
+    @Override
+    public boolean isValid(ProtocolTransportObject protocol) {
+        if (protocol == null) {
+            return false;
+        }
+        
+        // Check if it's an A2A message
+        try {
+            if (protocol instanceof RequestMessage) {
+                String content = protocol.toString();
+                return content.contains("\"protocol\":\"A2A\"") || 
+                       content.contains("A2A") ||
+                       content.contains("agent");
+            }
+            return false;
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
+    @Override
+    public CloudEvent toCloudEvent(ProtocolTransportObject protocolTransportObject) 
+        throws ProtocolHandleException {
         if (protocolTransportObject instanceof RequestMessage) {
             return toCloudEvent((RequestMessage) protocolTransportObject);
         } else if (protocolTransportObject instanceof ResponseMessage) {
             return toCloudEvent((ResponseMessage) protocolTransportObject);
         }
-        throw new IllegalArgumentException("Unsupported protocol transport object type");
+        throw new ProtocolHandleException("Unsupported protocol transport object type: " + 
+            protocolTransportObject.getClass().getName());
     }
 
     @Override
-    public ProtocolTransportObject fromCloudEvent(CloudEvent cloudEvent) {
+    public java.util.List<CloudEvent> toBatchCloudEvent(ProtocolTransportObject protocol) 
+        throws ProtocolHandleException {
+        // For A2A protocol, we can process multiple agent messages in batch
+        if (protocol instanceof RequestMessage) {
+            try {
+                String body = protocol.toString();
+                if (body.contains("batchMessages")) {
+                    // Parse batch A2A messages
+                    java.util.List<A2AMessage> batchMessages = parseBatchA2AMessages(body);
+                    return batchMessages.stream()
+                        .map(this::convertA2AMessageToCloudEvent)
+                        .collect(java.util.stream.Collectors.toList());
+                }
+            } catch (Exception e) {
+                throw new ProtocolHandleException("Failed to process batch A2A messages", e);
+            }
+        }
+        
+        // Fall back to single message processing
+        return java.util.List.of(toCloudEvent(protocol));
+    }
+
+    @Override
+    public ProtocolTransportObject fromCloudEvent(CloudEvent cloudEvent) 
+        throws ProtocolHandleException {
         String protocolType = cloudEvent.getAttributesMap().get("protocol").getCeString();
         if (!PROTOCOL_TYPE.equals(protocolType)) {
-            throw new IllegalArgumentException("Unsupported protocol type: " + protocolType);
+            throw new ProtocolHandleException("Unsupported protocol type: " + protocolType);
         }
 
         String messageType = cloudEvent.getAttributesMap().get("messageType").getCeString();
@@ -79,7 +173,7 @@ private CloudEvent toCloudEvent(RequestMessage requestMessage) {
 
             return cloudEventBuilder.build();
         } catch (Exception e) {
-            throw new RuntimeException("Failed to convert RequestMessage to CloudEvent", e);
+            throw new ProtocolHandleException("Failed to convert RequestMessage to CloudEvent", e);
         }
     }
 
@@ -102,7 +196,7 @@ private CloudEvent toCloudEvent(ResponseMessage responseMessage) {
 
             return cloudEventBuilder.build();
         } catch (Exception e) {
-            throw new RuntimeException("Failed to convert ResponseMessage to CloudEvent", e);
+            throw new ProtocolHandleException("Failed to convert ResponseMessage to CloudEvent", e);
         }
     }
 
@@ -111,7 +205,7 @@ private A2AMessage parseA2AMessage(CloudEvent cloudEvent) {
             String data = new String(cloudEvent.getData().toByteArray(), StandardCharsets.UTF_8);
             return JsonUtils.parseObject(data, A2AMessage.class);
         } catch (Exception e) {
-            throw new RuntimeException("Failed to parse A2A message from CloudEvent", e);
+            throw new ProtocolHandleException("Failed to parse A2A message from CloudEvent", e);
         }
     }
 
@@ -133,7 +227,73 @@ private ProtocolTransportObject createHttpMessage(A2AMessage a2aMessage) {
             RequestMessage requestMessage = new RequestMessage(header, body);
             return requestMessage;
         } catch (Exception e) {
-            throw new RuntimeException("Failed to create HTTP message from A2A message", e);
+            throw new ProtocolHandleException("Failed to create HTTP message from A2A message", e);
+        }
+    }
+
+    /**
+     * Parse batch A2A messages from request body.
+     */
+    private List<A2AMessage> parseBatchA2AMessages(String body) throws ProtocolHandleException {
+        try {
+            // Parse JSON to extract batch messages
+            Map<String, Object> bodyMap = JsonUtils.parseTypeReferenceObject(body, 
+                new com.fasterxml.jackson.core.type.TypeReference<Map<String, Object>>() {});
+            
+            if (bodyMap.containsKey("batchMessages")) {
+                @SuppressWarnings("unchecked")
+                List<Map<String, Object>> messagesData = (List<Map<String, Object>>) bodyMap.get("batchMessages");
+                
+                return messagesData.stream()
+                    .map(data -> JsonUtils.mapToObject(data, A2AMessage.class))
+                    .collect(java.util.stream.Collectors.toList());
+            }
+            
+            return List.of();
+        } catch (Exception e) {
+            throw new ProtocolHandleException("Failed to parse batch A2A messages", e);
+        }
+    }
+
+    /**
+     * Convert A2A message to CloudEvent.
+     */
+    private CloudEvent convertA2AMessageToCloudEvent(A2AMessage a2aMessage) {
+        try {
+            CloudEvent.Builder cloudEventBuilder = CloudEvent.newBuilder()
+                .setId(a2aMessage.getMessageId() != null ? a2aMessage.getMessageId() : UUID.randomUUID().toString())
+                .setSource("eventmesh-a2a")
+                .setSpecVersion("1.0")
+                .setType("org.apache.eventmesh.protocol.a2a." + a2aMessage.getMessageType().toLowerCase())
+                .setData(com.google.protobuf.ByteString.copyFromUtf8(JsonUtils.toJSONString(a2aMessage)))
+                .putAttributes("protocol", 
+                    CloudEvent.CloudEventAttributeValue.newBuilder().setCeString(PROTOCOL_TYPE).build())
+                .putAttributes("version", 
+                    CloudEvent.CloudEventAttributeValue.newBuilder().setCeString(PROTOCOL_VERSION).build())
+                .putAttributes("messageType", 
+                    CloudEvent.CloudEventAttributeValue.newBuilder().setCeString(a2aMessage.getMessageType()).build());
+
+            if (a2aMessage.getSourceAgent() != null) {
+                cloudEventBuilder.putAttributes("sourceAgent", 
+                    CloudEvent.CloudEventAttributeValue.newBuilder()
+                        .setCeString(a2aMessage.getSourceAgent().getAgentId()).build());
+            }
+
+            if (a2aMessage.getTargetAgent() != null) {
+                cloudEventBuilder.putAttributes("targetAgent", 
+                    CloudEvent.CloudEventAttributeValue.newBuilder()
+                        .setCeString(a2aMessage.getTargetAgent().getAgentId()).build());
+            }
+
+            if (a2aMessage.getMetadata() != null && a2aMessage.getMetadata().getCorrelationId() != null) {
+                cloudEventBuilder.putAttributes("correlationId", 
+                    CloudEvent.CloudEventAttributeValue.newBuilder()
+                        .setCeString(a2aMessage.getMetadata().getCorrelationId()).build());
+            }
+
+            return cloudEventBuilder.build();
+        } catch (Exception e) {
+            throw new RuntimeException("Failed to convert A2A message to CloudEvent", e);
         }
     }