You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2007/01/10 21:12:25 UTC
svn commit: r494962 - in /incubator/qpid/branches/qpid.0-9/java:
broker/src/main/java/org/apache/qpid/server/
common/src/main/java/org/apache/qpid/framing/
Author: kpvdr
Date: Wed Jan 10 12:12:23 2007
New Revision: 494962
URL: http://svn.apache.org/viewvc?view=rev&rev=494962
Log:
First version of RequestResponseHandler for an idea of how the RequestResponse frames will be managed... Not quite complete, still need to find a means of writing the frames that is common to both client and server.
Added:
incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/RequestResponseManager.java
Modified:
incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQRequestBody.java
incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQResponseBody.java
Modified: incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?view=diff&rev=494962&r1=494961&r2=494962
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java Wed Jan 10 12:12:23 2007
@@ -170,49 +170,48 @@
_prefetch_HighWaterMark = prefetchCount;
}
-
public void setPublishFrame(BasicPublishBody publishBody, AMQProtocolSession publisher) throws AMQException
{
_currentMessage = new AMQMessage(_messageStore, publishBody);
_currentMessage.setPublisher(publisher);
}
- public void publishContentHeader(ContentHeaderBody contentHeaderBody)
- throws AMQException
- {
- if (_currentMessage == null)
- {
- throw new AMQException("Received content header without previously receiving a BasicDeliver frame");
- }
- else
- {
- _currentMessage.setContentHeaderBody(contentHeaderBody);
- // check and route if header says body length is zero
- if (contentHeaderBody.bodySize == 0)
- {
- routeCurrentMessage();
- }
- }
- }
-
- public void publishContentBody(ContentBody contentBody)
- throws AMQException
- {
- if (_currentMessage == null)
- {
- throw new AMQException("Received content body without previously receiving a JmsPublishBody");
- }
- if (_currentMessage.getContentHeaderBody() == null)
- {
- throw new AMQException("Received content body without previously receiving a content header");
- }
-
- _currentMessage.addContentBodyFrame(contentBody);
- if (_currentMessage.isAllContentReceived())
- {
- routeCurrentMessage();
- }
- }
+// public void publishContentHeader(ContentHeaderBody contentHeaderBody)
+// throws AMQException
+// {
+// if (_currentMessage == null)
+// {
+// throw new AMQException("Received content header without previously receiving a BasicDeliver frame");
+// }
+// else
+// {
+// _currentMessage.setContentHeaderBody(contentHeaderBody);
+// // check and route if header says body length is zero
+// if (contentHeaderBody.bodySize == 0)
+// {
+// routeCurrentMessage();
+// }
+// }
+// }
+//
+// public void publishContentBody(ContentBody contentBody)
+// throws AMQException
+// {
+// if (_currentMessage == null)
+// {
+// throw new AMQException("Received content body without previously receiving a JmsPublishBody");
+// }
+// if (_currentMessage.getContentHeaderBody() == null)
+// {
+// throw new AMQException("Received content body without previously receiving a content header");
+// }
+//
+// _currentMessage.addContentBodyFrame(contentBody);
+// if (_currentMessage.isAllContentReceived())
+// {
+// routeCurrentMessage();
+// }
+// }
protected void routeCurrentMessage() throws AMQException
{
Modified: incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQRequestBody.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQRequestBody.java?view=diff&rev=494962&r1=494961&r2=494962
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQRequestBody.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQRequestBody.java Wed Jan 10 12:12:23 2007
@@ -24,16 +24,22 @@
public class AMQRequestBody extends AMQBody
{
- public static final byte TYPE = (byte)AmqpConstants.frameRequestAsInt();
-
// Fields declared in specification
- public long requestId;
- public long responseMark;
- public AMQMethodBody methodPayload;
+ protected long requestId;
+ protected long responseMark;
+ protected AMQMethodBody methodPayload;
// Constructor
public AMQRequestBody() {}
+ public AMQRequestBody(long requestId, long responseMark,
+ AMQMethodBody methodPayload)
+ {
+ this.requestId = requestId;
+ this.responseMark = responseMark;
+ this.methodPayload = methodPayload;
+ }
+
// Field methods
public long getRequestId() { return requestId; }
@@ -43,7 +49,7 @@
protected byte getFrameType()
{
- return TYPE;
+ return (byte)AmqpConstants.frameRequestAsInt();
}
protected int getSize()
@@ -68,15 +74,17 @@
methodPayload.populateFromBuffer(buffer, size - 8 - 8 - 4);
}
+ public String toString()
+ {
+ return "Req[" + requestId + " " + responseMark + "] C" +
+ methodPayload.getClazz() + " M" + methodPayload.getMethod();
+ }
+
public static AMQFrame createAMQFrame(int channelId, long requestId,
long responseMark, AMQMethodBody methodPayload)
{
- AMQRequestBody requestFrame = new AMQRequestBody();
- requestFrame.requestId = requestId;
- requestFrame.responseMark = responseMark;
- requestFrame.methodPayload = methodPayload;
-
-
+ AMQRequestBody requestFrame = new AMQRequestBody(requestId, responseMark,
+ methodPayload);
AMQFrame frame = new AMQFrame();
frame.channel = channelId;
frame.bodyFrame = requestFrame;
Modified: incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQResponseBody.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQResponseBody.java?view=diff&rev=494962&r1=494961&r2=494962
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQResponseBody.java (original)
+++ incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/AMQResponseBody.java Wed Jan 10 12:12:23 2007
@@ -24,16 +24,22 @@
public class AMQResponseBody extends AMQBody
{
- public static final byte TYPE = (byte)AmqpConstants.frameResponseAsInt();
-
// Fields declared in specification
- public long responseId;
- public long requestId;
- public int batchOffset;
- public AMQMethodBody methodPayload;
+ protected long responseId;
+ protected long requestId;
+ protected int batchOffset;
+ protected AMQMethodBody methodPayload;
// Constructor
public AMQResponseBody() {}
+ public AMQResponseBody(long getResponseId, long getRequestId,
+ int batchOffset, AMQMethodBody methodPayload)
+ {
+ this.responseId = responseId;
+ this.requestId = requestId;
+ this.batchOffset = batchOffset;
+ this.methodPayload = methodPayload;
+ }
// Field methods
public long getResponseId() { return responseId; }
@@ -43,7 +49,7 @@
protected byte getFrameType()
{
- return TYPE;
+ return (byte)AmqpConstants.frameResponseAsInt();
}
protected int getSize()
@@ -68,15 +74,17 @@
methodPayload.populateFromBuffer(buffer, size - 8 - 8 - 4);
}
- public static AMQFrame createAMQFrame(int channelId, long requestId,
- long responseId, int batchOffset, AMQMethodBody methodPayload)
+ public String toString()
{
- AMQResponseBody responseFrame = new AMQResponseBody();
- responseFrame.responseId = responseId;
- responseFrame.requestId = requestId;
- responseFrame.batchOffset = batchOffset;
- responseFrame.methodPayload = methodPayload;
-
+ return "Res[" + responseId + " " + requestId + "-" + requestId + batchOffset + "] C" +
+ methodPayload.getClazz() + " M" + methodPayload.getMethod();
+ }
+
+ public static AMQFrame createAMQFrame(int channelId, long responseId,
+ long requestId, int batchOffset, AMQMethodBody methodPayload)
+ {
+ AMQResponseBody responseFrame = new AMQResponseBody(responseId,
+ requestId, batchOffset, methodPayload);
AMQFrame frame = new AMQFrame();
frame.channel = channelId;
frame.bodyFrame = responseFrame;
Added: incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/RequestResponseManager.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/RequestResponseManager.java?view=auto&rev=494962
==============================================================================
--- incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/RequestResponseManager.java (added)
+++ incubator/qpid/branches/qpid.0-9/java/common/src/main/java/org/apache/qpid/framing/RequestResponseManager.java Wed Jan 10 12:12:23 2007
@@ -0,0 +1,224 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.framing;
+
+import java.lang.reflect.Method;
+import java.util.Iterator;
+import java.util.TreeMap;
+
+import org.apache.qpid.AMQException;
+//import org.apache.qpid.server.protocol.AMQProtocolSession;
+
+public class RequestResponseManager
+{
+ private int channel;
+// AMQProtocolSession protocolSession;
+
+ /**
+ * Determines the batch behaviour of the manager.
+ *
+ * Responses are sent to the RequestResponseManager through sendResponse().
+ * These may be internally stored/accumulated for batching purposes, depending
+ * on the batching strategy/mode of the RequestResponseManager.
+ *
+ * The following modes are possibe:
+ *
+ * NONE: Each request results in an immediate single response, no batching
+ * takes place.
+ * DELAY_FIXED: Waits until a fixed period has passed to batch
+ * accumulated responses. An optional fixed threshold may be set, which
+ * if reached or exceeded within the delay period will trigger the batch. (TODO)
+ * MANUAL: No response is sent until it is explicitly released by calling
+ * function xxxx(). (TODO)
+ */
+ public enum batchResponseModeEnum { NONE }
+ private batchResponseModeEnum batchResponseMode;
+
+ /**
+ * Request and response frames must have a requestID and responseID which
+ * indepenedently increment from 0 on a per-channel basis. These are the
+ * counters, and contain the value of the next (not yet used) frame.
+ */
+ private long requestIdCount;
+ private long responseIdCount;
+
+ /**
+ * These keep track of the last requestId and responseId to be received.
+ */
+ private long lastReceivedRequestId;
+ private long lastReceivedResponseId;
+
+ /**
+ * Last requestID sent in a response (for batching)
+ */
+ private long lastSentRequestId;
+
+ private class ResponseStatus implements Comparable<ResponseStatus>
+ {
+ public long requestId;
+ public AMQMethodBody responseMethodBody;
+
+ public ResponseStatus(long requestId)
+ {
+ this.requestId = requestId;
+ responseMethodBody = null;
+ }
+
+ public int compareTo(ResponseStatus o)
+ {
+ return (int)(requestId - o.requestId);
+ }
+ }
+
+ private TreeMap<Long, Method> requestSentMap;
+ private TreeMap<Long, ResponseStatus> responseMap;
+
+// public RequestResponseManager(int channel, AMQProtocolSession protocolSession)
+ public RequestResponseManager(int channel)
+ {
+ this.channel = channel;
+// this.protocolSession = protocolSession;
+ requestIdCount = 1L;
+ responseIdCount = 1L;
+ lastReceivedRequestId = 0L;
+ lastReceivedResponseId = 0L;
+ requestSentMap = new TreeMap<Long, Method>();
+ responseMap = new TreeMap<Long, ResponseStatus>();
+ }
+
+ // *** Functions to originate a request ***
+
+ public long sendRequest(AMQMethodBody requestMethodBody, Method responseCallback)
+ {
+ long requestId = getRequestId(); // Get new request ID
+ AMQFrame requestFrame = AMQRequestBody.createAMQFrame(channel, requestId,
+ lastReceivedResponseId, requestMethodBody);
+// protocolSession.writeFrame(requestFrame);
+ requestSentMap.put(requestId, responseCallback);
+ return requestId;
+ }
+
+ public void responseReceived(AMQResponseBody responseBody) throws AMQException
+ {
+ lastReceivedResponseId = responseBody.getResponseId();
+ long requestIdStart = responseBody.getRequestId();
+ long requestIdStop = requestIdStart + responseBody.getBatchOffset();
+ for (long requestId = requestIdStart; requestId <= requestIdStop; requestId++)
+ {
+ Method responseCallback = requestSentMap.get(requestId);
+ if (responseCallback == null)
+ throw new AMQException("Failed to locate requestId " + requestId +
+ " in requestSentMap.");
+ // TODO
+ // responseCallback.invoke(?);
+ requestSentMap.remove(requestId);
+ }
+ }
+
+ // *** Functions to handle an incoming request ***
+
+ public void requestReceived(AMQRequestBody requestBody)
+ {
+ long requestId = requestBody.getRequestId();
+ long responseMark = requestBody.getResponseMark(); // TODO - what do we do with this??
+ lastReceivedRequestId = requestId;
+ responseMap.put(requestId, new ResponseStatus(requestId));
+
+ // TODO: Initiate some action based on the MethodBody
+ }
+
+ public void sendResponse(long requestId, AMQMethodBody responseMethodBody) throws AMQException
+ {
+ ResponseStatus responseStatus = responseMap.get(requestId);
+ if (responseStatus == null)
+ throw new AMQException("Failed to locate requestId " + requestId +
+ " in responseMap.");
+ if (responseStatus.responseMethodBody != null)
+ throw new AMQException("RequestId " + requestId + " already has a response.");
+ responseStatus.responseMethodBody = responseMethodBody;
+ doBatches();
+ }
+
+ // *** Management functions ***
+
+ public batchResponseModeEnum getBatchResponseMode()
+ {
+ return batchResponseMode;
+ }
+
+ public void setBatchResponseMode(batchResponseModeEnum batchResponseMode)
+ {
+ if (this.batchResponseMode != batchResponseMode)
+ {
+ this.batchResponseMode = batchResponseMode;
+ doBatches();
+ }
+ }
+
+ // *** Private helper functions ***
+
+ private long getRequestId()
+ {
+ return requestIdCount++;
+ }
+
+ private long getResponseId()
+ {
+ return responseIdCount++;
+ }
+
+/* private Method findRequest(long requestId)
+ throws AMQException
+ {
+ RequestStatus requestStatus = requestMap.get(requestId);
+ // TODO
+ return null;
+ }
+*/
+ private void doBatches()
+ {
+ switch (batchResponseMode)
+ {
+ case NONE:
+ Iterator<Long> lItr = responseMap.keySet().iterator();
+ while (lItr.hasNext())
+ {
+ long requestId = lItr.next();
+ ResponseStatus responseStatus = responseMap.get(requestId);
+ if (responseStatus.responseMethodBody != null)
+ {
+ sendResponseBatch(requestId, 0, responseStatus.responseMethodBody);
+ lItr.remove();
+ }
+ }
+ break;
+ }
+ }
+
+ private void sendResponseBatch(long firstRequestId, int numAdditionalRequests,
+ AMQMethodBody responseMethodBody)
+ {
+ long responseId = getResponseId(); // Get new request ID
+ AMQFrame responseFrame = AMQResponseBody.createAMQFrame(channel, responseId,
+ firstRequestId, numAdditionalRequests, responseMethodBody);
+// protocolSession.writeFrame(responseFrame);
+ }
+}