Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Tidied up the async dispatch option 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@494950 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2007-01-10 19:40:33 +00:00
parent b21113962b
commit f19a3a66d6
1 changed files with 28 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java
View File

@ -11,6 +11,7 @@
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * 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. * specific language governing permissions and limitations under the License.
*/ */
package org.apache.activemq.transport.vm; package org.apache.activemq.transport.vm;
import java.io.IOException; import java.io.IOException;
@ -19,6 +20,7 @@ import java.util.Collections;
import java.util.Iterator; import java.util.Iterator;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List; import java.util.List;
import org.apache.activemq.broker.BrokerStoppedException;
import org.apache.activemq.command.Command; import org.apache.activemq.command.Command;
import org.apache.activemq.thread.Task; import org.apache.activemq.thread.Task;
import org.apache.activemq.thread.TaskRunner; import org.apache.activemq.thread.TaskRunner;
@ -32,16 +34,18 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
/** /**
* A Transport implementation that uses direct method invocations. * A Transport implementation that uses direct method invocations.
* *
* @version $Revision$ * @version $Revision$
*/ */
public class VMTransport implements Transport,Task{ public class VMTransport implements Transport,Task{
private static final Log log=LogFactory.getLog(VMTransport.class); private static final Log log=LogFactory.getLog(VMTransport.class);
private static final AtomicLong nextId=new AtomicLong(0); private static final AtomicLong nextId=new AtomicLong(0);
private static final TaskRunnerFactory taskRunnerFactory=new TaskRunnerFactory("VMTransport",Thread.NORM_PRIORITY, private static final TaskRunnerFactory taskRunnerFactory=new TaskRunnerFactory("VMTransport",Thread.NORM_PRIORITY,
true,1000); true,1000);
protected VMTransport peer; protected VMTransport peer;
protected TransportListener transportListener; protected TransportListener transportListener;
protected boolean disposed; protected boolean disposed;
@ -51,7 +55,7 @@ public class VMTransport implements Transport,Task{
protected boolean started=false; protected boolean started=false;
protected int asyncQueueDepth=2000; protected int asyncQueueDepth=2000;
protected List prePeerSetQueue=Collections.synchronizedList(new LinkedList()); protected List prePeerSetQueue=Collections.synchronizedList(new LinkedList());
protected LinkedBlockingQueue messageQueue; protected LinkedBlockingQueue messageQueue=null;
protected final URI location; protected final URI location;
protected final long id; protected final long id;
private TaskRunner taskRunner; private TaskRunner taskRunner;
@ -76,9 +80,8 @@ public class VMTransport implements Transport,Task{
if(peer==null) if(peer==null)
throw new IOException("Peer not connected."); throw new IOException("Peer not connected.");
if(!peer.disposed){ if(!peer.disposed){
if(async){ if(async){
asyncOneWay(command); asyncOneWay(command);
}else{ }else{
syncOneWay(command); syncOneWay(command);
} }
@ -86,7 +89,7 @@ public class VMTransport implements Transport,Task{
throw new TransportDisposedIOException("Peer ("+peer.toString()+") disposed."); throw new TransportDisposedIOException("Peer ("+peer.toString()+") disposed.");
} }
} }
protected void syncOneWay(Object command){ protected void syncOneWay(Object command){
final TransportListener tl=peer.transportListener; final TransportListener tl=peer.transportListener;
prePeerSetQueue=peer.prePeerSetQueue; prePeerSetQueue=peer.prePeerSetQueue;
@ -96,10 +99,12 @@ public class VMTransport implements Transport,Task{
tl.onCommand(command); tl.onCommand(command);
} }
} }
protected void asyncOneWay(Object command) throws IOException{ protected synchronized void asyncOneWay(Object command) throws IOException{
messageQueue=getMessageQueue();
try{ try{
if(messageQueue==null){
messageQueue=new LinkedBlockingQueue(this.asyncQueueDepth);
}
messageQueue.put(command); messageQueue.put(command);
wakeup(); wakeup();
}catch(final InterruptedException e){ }catch(final InterruptedException e){
@ -136,17 +141,17 @@ public class VMTransport implements Transport,Task{
throw new IOException("TransportListener not set."); throw new IOException("TransportListener not set.");
if(!async){ if(!async){
for(Iterator iter=prePeerSetQueue.iterator();iter.hasNext();){ for(Iterator iter=prePeerSetQueue.iterator();iter.hasNext();){
Command command=(Command) iter.next(); Command command=(Command)iter.next();
transportListener.onCommand(command); transportListener.onCommand(command);
iter.remove(); iter.remove();
} }
}else{ }else{
wakeup();
peer.wakeup(); peer.wakeup();
wakeup();
} }
} }
public void stop() throws Exception{ public synchronized void stop() throws Exception{
started=false; started=false;
if(!disposed){ if(!disposed){
disposed=true; disposed=true;
@ -196,11 +201,17 @@ public class VMTransport implements Transport,Task{
*/ */
public boolean iterate(){ public boolean iterate(){
final TransportListener tl=peer.transportListener; final TransportListener tl=peer.transportListener;
if(!messageQueue.isEmpty()&&!peer.disposed&&tl!=null){ Command command=null;
final Command command=(Command) messageQueue.poll(); // if(!disposed && !messageQueue.isEmpty()&&!peer.disposed&&tl!=null){
tl.onCommand(command); synchronized(this){
if(messageQueue!=null&&!disposed&&!peer.disposed&&tl!=null &&!messageQueue.isEmpty()){
command=(Command)messageQueue.poll();
if (command != null) {
tl.onCommand(command);
}
}
} }
return !messageQueue.isEmpty()&&!peer.disposed&&!(peer.transportListener==null); return messageQueue!=null&&!messageQueue.isEmpty()&&!peer.disposed;
} }
/** /**
@ -231,8 +242,8 @@ public class VMTransport implements Transport,Task{
this.asyncQueueDepth=asyncQueueDepth; this.asyncQueueDepth=asyncQueueDepth;
} }
protected void wakeup(){ protected synchronized void wakeup(){
if(async&&messageQueue!=null&&!messageQueue.isEmpty()){ if(async){
if(taskRunner==null){ if(taskRunner==null){
taskRunner=taskRunnerFactory.createTaskRunner(this,"VMTransport: "+toString()); taskRunner=taskRunnerFactory.createTaskRunner(this,"VMTransport: "+toString());
} }
@ -243,11 +254,4 @@ public class VMTransport implements Transport,Task{
} }
} }
} }
protected synchronized LinkedBlockingQueue getMessageQueue(){
if(messageQueue==null){
messageQueue=new LinkedBlockingQueue(this.asyncQueueDepth);
}
return messageQueue;
}
} }