9:43 a.m.
Author: richard.opalka(a)jboss.com
Date: 2007-12-05 10:43:57 -0500 (Wed, 05 Dec 2007)
New Revision: 5193
Added:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManager.java
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManagerImpl.java
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelTask.java
Removed:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelRequest.java
Modified:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/RMSequenceImpl.java
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannel.java
Log:
refactoring - all WS-RM messages will go through RMChannelManager
Modified:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/RMSequenceImpl.java
===================================================================
---
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/RMSequenceImpl.java 2007-12-05
15:25:26 UTC (rev 5192)
+++
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/RMSequenceImpl.java 2007-12-05
15:43:57 UTC (rev 5193)
@@ -25,7 +25,6 @@
import java.net.URI;
import java.net.URISyntaxException;
import java.net.UnknownHostException;
-import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
@@ -66,6 +65,7 @@
public final class RMSequenceImpl implements RMSequence, RMUnassignedMessageListener
{
private static final Logger logger = Logger.getLogger(RMSequenceImpl.class);
+ private static final String PATH_PREFIX = "/temporary_listen_address/";
private static final RMConstants wsrmConstants = RMProvider.get().getConstants();
private final RMConfig wsrmConfig;
@@ -111,7 +111,7 @@
logger.debug("Backports server configuration omits host configuration -
using autodetected " + host);
}
String port = wsrmConfig.getBackPortsServer().getPort();
- String path = "/temporary_listen_address/" +
UUIDGenerator.generateRandomUUIDString();
+ String path = PATH_PREFIX + UUIDGenerator.generateRandomUUIDString();
this.backPort = new URI("http://" + host + ":" + port +
path);
}
catch (URISyntaxException use)
@@ -128,7 +128,7 @@
public final Set<Long> getReceivedInboundMessages()
{
- return Collections.unmodifiableSet(this.receivedInboundMessages);
+ return this.receivedInboundMessages;
}
public final BindingProvider getBindingProvider()
Modified:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannel.java
===================================================================
---
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannel.java 2007-12-05
15:25:26 UTC (rev 5192)
+++
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannel.java 2007-12-05
15:43:57 UTC (rev 5193)
@@ -32,10 +32,6 @@
import java.io.ByteArrayOutputStream;
import java.util.Map;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.atomic.AtomicInteger;
/**
* RM Channel
@@ -56,30 +52,8 @@
}
// Holds the list of tasks that will be send to the remoting transport channel
- private static final ExecutorService rmChannelPool = Executors.newFixedThreadPool(5,
new RMThreadFactory());
+ private static final RMChannelManager rmChannelManager =
RMChannelManagerImpl.getInstance();
- private static final class RMThreadFactory implements ThreadFactory
- {
- final ThreadGroup group;
- final AtomicInteger threadNumber = new AtomicInteger(1);
- final String namePrefix = "rm-pool-thread-";
-
- private RMThreadFactory()
- {
- SecurityManager sm = System.getSecurityManager();
- group = (sm != null) ? sm.getThreadGroup() :
Thread.currentThread().getThreadGroup();
- }
-
- public Thread newThread(Runnable r)
- {
- Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(),
0);
- if (t.isDaemon())
- t.setDaemon(false);
- if (t.getPriority() != Thread.NORM_PRIORITY)
- t.setPriority(Thread.NORM_PRIORITY);
- return t;
- }
- }
private RMMessage createRMMessage(MessageAbstraction request, RMMetadata rmMetadata)
throws Throwable
{
@@ -122,16 +96,7 @@
private RMMessage sendToChannel(RMMessage request) throws Throwable
{
- RMChannelResponse result = rmChannelPool.submit(new
RMChannelRequest(request)).get();
-
- Throwable fault = result.getFault();
- if (fault != null)
- {
- throw fault;
- }
- else
- {
- return result.getResponse();
- }
+ return rmChannelManager.send(request);
}
+
}
Added:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManager.java
===================================================================
---
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManager.java
(rev 0)
+++
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManager.java 2007-12-05
15:43:57 UTC (rev 5193)
@@ -0,0 +1,41 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005, JBoss Inc., and individual contributors as indicated
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.ws.extensions.wsrm.transport;
+
+/**
+ * WS-RM channel manager interface
+ *
+ * @author richard.opalka(a)jboss.com
+ *
+ * @since Dec 5, 2007
+ */
+public interface RMChannelManager
+{
+
+ /**
+ * Send request to server reliably
+ * @param request RM request wrapper
+ * @return RM response wrapper
+ */
+ RMMessage send(RMMessage request) throws Throwable;
+
+}
Property changes on:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManager.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Added:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManagerImpl.java
===================================================================
---
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManagerImpl.java
(rev 0)
+++
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManagerImpl.java 2007-12-05
15:43:57 UTC (rev 5193)
@@ -0,0 +1,112 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005, JBoss Inc., and individual contributors as indicated
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.ws.extensions.wsrm.transport;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.jboss.logging.Logger;
+
+/**
+ * WS-RM channel manager ensures message reliable delivery according to sequence
configuration
+ *
+ * @author richard.opalka(a)jboss.com
+ *
+ * @since Dec 5, 2007
+ */
+public final class RMChannelManagerImpl implements Runnable, RMChannelManager
+{
+
+ private static final Logger logger = Logger.getLogger(RMChannelManagerImpl.class);
+ private static RMChannelManager instance = new RMChannelManagerImpl();
+ private static final ExecutorService rmChannelPool = Executors.newFixedThreadPool(5,
new RMThreadFactory());
+
+ private static final class RMThreadFactory implements ThreadFactory
+ {
+ final ThreadGroup group;
+ final AtomicInteger threadNumber = new AtomicInteger(1);
+ final String namePrefix = "rm-pool-thread-";
+
+ private RMThreadFactory()
+ {
+ SecurityManager sm = System.getSecurityManager();
+ group = (sm != null) ? sm.getThreadGroup() :
Thread.currentThread().getThreadGroup();
+ }
+
+ public Thread newThread(Runnable r)
+ {
+ Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(),
0);
+ if (t.isDaemon())
+ t.setDaemon(false);
+ if (t.getPriority() != Thread.NORM_PRIORITY)
+ t.setPriority(Thread.NORM_PRIORITY);
+ return t;
+ }
+ }
+
+ private RMChannelManagerImpl()
+ {
+ Thread thread = new Thread(this, "RMChannelManager");
+ thread.setDaemon(true);
+ thread.start();
+ }
+
+ public static final RMChannelManager getInstance()
+ {
+ return instance;
+ }
+
+ public final void run()
+ {
+ while (true)
+ {
+ logger.debug("checking persistent store for undelivered messages");
+ try
+ {
+ Thread.sleep(10);
+ }
+ catch (InterruptedException ie)
+ {
+ logger.warn(ie.getMessage(), ie);
+ }
+ }
+
+ }
+
+ public final RMMessage send(RMMessage request) throws Throwable
+ {
+ RMChannelResponse result = rmChannelPool.submit(new RMChannelTask(request)).get();
+
+ Throwable fault = result.getFault();
+ if (fault != null)
+ {
+ throw fault;
+ }
+ else
+ {
+ return result.getResponse();
+ }
+ }
+
+}
Property changes on:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelManagerImpl.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF
Deleted:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelRequest.java
===================================================================
---
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelRequest.java 2007-12-05
15:25:26 UTC (rev 5192)
+++
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelRequest.java 2007-12-05
15:43:57 UTC (rev 5193)
@@ -1,143 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005, JBoss Inc., and individual contributors as indicated
- * by the @authors tag. See the copyright.txt in the distribution for a
- * full listing of individual contributors.
- *
- * This is free software; you can redistribute it and/or modify it
- * under the terms of the GNU Lesser General Public License as
- * published by the Free Software Foundation; either version 2.1 of
- * the License, or (at your option) any later version.
- *
- * This software is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this software; if not, write to the Free
- * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
- * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
- */
-package org.jboss.ws.extensions.wsrm.transport;
-
-import static org.jboss.ws.extensions.wsrm.RMConstant.*;
-
-import java.net.MalformedURLException;
-import java.net.URI;
-import java.util.Map;
-import java.util.concurrent.Callable;
-
-import org.jboss.logging.Logger;
-import org.jboss.remoting.Client;
-import org.jboss.remoting.InvokerLocator;
-import org.jboss.ws.core.MessageTrace;
-import org.jboss.ws.extensions.wsrm.RMSequenceImpl;
-import org.jboss.ws.extensions.wsrm.transport.backchannel.RMCallbackHandler;
-import org.jboss.ws.extensions.wsrm.transport.backchannel.RMCallbackHandlerFactory;
-
-/**
- * Represents request that goes to the RM channel
- * @see org.jboss.ws.extensions.wsrm.transport.RMChannel
- * @author richard.opalka(a)jboss.com
- */
-public final class RMChannelRequest implements Callable<RMChannelResponse>
-{
- private static final Logger logger = Logger.getLogger(RMChannelRequest.class);
- private static final String JBOSSWS_SUBSYSTEM = "jbossws";
- private final RMMessage rmRequest;
-
- RMChannelRequest(RMMessage rmRequest)
- {
- super();
- this.rmRequest = rmRequest;
- }
-
- public RMChannelResponse call()
- {
- InvokerLocator locator = null;
- try
- {
- locator = new
InvokerLocator((String)rmRequest.getMetadata().getContext(INVOCATION_CONTEXT).get(TARGET_ADDRESS));
- }
- catch (MalformedURLException e)
- {
- return new RMChannelResponse(new IllegalArgumentException("Malformed
endpoint address", e));
- }
-
- try
- {
- URI backPort = RMTransportHelper.getBackPortURI(rmRequest);
- String messageId = RMTransportHelper.getMessageId(rmRequest);
-
- logger.debug("[WS-RM] backport URI is: " + backPort);
- RMCallbackHandler callbackHandler = null;
- // TODO: we should remember WSA:MessageId here too
-
- if (backPort != null)
- {
- callbackHandler = RMCallbackHandlerFactory.getCallbackHandler(backPort);
- RMSequenceImpl sequence = RMTransportHelper.getSequence(rmRequest);
- if (sequence != null)
- {
- callbackHandler.addUnassignedMessageListener(sequence);
- }
- }
- boolean oneWay = RMTransportHelper.isOneWayOperation(rmRequest);
-
- Client client = new Client(locator, JBOSSWS_SUBSYSTEM,
rmRequest.getMetadata().getContext(REMOTING_CONFIGURATION_CONTEXT));
- client.connect();
-
- client.setMarshaller(RMMarshaller.getInstance());
-
- if ((false == oneWay) && (null == backPort))
- client.setUnMarshaller(RMUnMarshaller.getInstance());
-
- Map<String, Object> remotingInvocationContext =
rmRequest.getMetadata().getContext(REMOTING_INVOCATION_CONTEXT);
- if (logger.isDebugEnabled())
- logger.debug("Remoting metadata: " + remotingInvocationContext);
-
- // debug the outgoing request message
- MessageTrace.traceMessage("Outgoing RM Request Message",
rmRequest.getPayload());
-
- RMMessage rmResponse = null;
- if (oneWay && (null == backPort))
- {
- client.invokeOneway(rmRequest.getPayload(), remotingInvocationContext,
false);
- }
- else
- {
- Object retVal = client.invoke(rmRequest.getPayload(),
remotingInvocationContext);
- if ((null != retVal) && (false == (retVal instanceof RMMessage)))
- {
- String msg = retVal.getClass().getName() + ": '" + retVal +
"'";
- logger.warn(msg);
- throw new RuntimeException(msg);
- }
- rmResponse = (RMMessage)retVal;
- }
-
- // Disconnect the remoting client
- client.disconnect();
-
- // trace the incomming response message
- if (rmResponse != null)
- MessageTrace.traceMessage("Incoming RM Response Message",
rmResponse.getPayload());
-
- if (backPort != null) // TODO: backport support
- {
- if ((null != messageId) && (false ==
RMTransportHelper.isOneWayOperation(rmRequest)))
- {
- // register callbacks only for outbound messages with messageId
- return new RMChannelResponse(callbackHandler, messageId);
- }
- }
-
- return new RMChannelResponse(rmResponse);
- }
- catch (Throwable t)
- {
- return new RMChannelResponse(t);
- }
- }
-}
\ No newline at end of file
Added:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelTask.java
===================================================================
---
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelTask.java
(rev 0)
+++
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelTask.java 2007-12-05
15:43:57 UTC (rev 5193)
@@ -0,0 +1,143 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005, JBoss Inc., and individual contributors as indicated
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.jboss.ws.extensions.wsrm.transport;
+
+import static org.jboss.ws.extensions.wsrm.RMConstant.*;
+
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import org.jboss.logging.Logger;
+import org.jboss.remoting.Client;
+import org.jboss.remoting.InvokerLocator;
+import org.jboss.ws.core.MessageTrace;
+import org.jboss.ws.extensions.wsrm.RMSequenceImpl;
+import org.jboss.ws.extensions.wsrm.transport.backchannel.RMCallbackHandler;
+import org.jboss.ws.extensions.wsrm.transport.backchannel.RMCallbackHandlerFactory;
+
+/**
+ * Represents request that goes to the RM channel
+ * @see org.jboss.ws.extensions.wsrm.transport.RMChannel
+ * @author richard.opalka(a)jboss.com
+ */
+public final class RMChannelTask implements Callable<RMChannelResponse>
+{
+ private static final Logger logger = Logger.getLogger(RMChannelTask.class);
+ private static final String JBOSSWS_SUBSYSTEM = "jbossws";
+ private final RMMessage rmRequest;
+
+ RMChannelTask(RMMessage rmRequest)
+ {
+ super();
+ this.rmRequest = rmRequest;
+ }
+
+ public RMChannelResponse call()
+ {
+ InvokerLocator locator = null;
+ try
+ {
+ locator = new
InvokerLocator((String)rmRequest.getMetadata().getContext(INVOCATION_CONTEXT).get(TARGET_ADDRESS));
+ }
+ catch (MalformedURLException e)
+ {
+ return new RMChannelResponse(new IllegalArgumentException("Malformed
endpoint address", e));
+ }
+
+ try
+ {
+ URI backPort = RMTransportHelper.getBackPortURI(rmRequest);
+ String messageId = RMTransportHelper.getMessageId(rmRequest);
+
+ logger.debug("[WS-RM] backport URI is: " + backPort);
+ RMCallbackHandler callbackHandler = null;
+ // TODO: we should remember WSA:MessageId here too
+
+ if (backPort != null)
+ {
+ callbackHandler = RMCallbackHandlerFactory.getCallbackHandler(backPort);
+ RMSequenceImpl sequence = RMTransportHelper.getSequence(rmRequest);
+ if (sequence != null)
+ {
+ callbackHandler.addUnassignedMessageListener(sequence);
+ }
+ }
+ boolean oneWay = RMTransportHelper.isOneWayOperation(rmRequest);
+
+ Client client = new Client(locator, JBOSSWS_SUBSYSTEM,
rmRequest.getMetadata().getContext(REMOTING_CONFIGURATION_CONTEXT));
+ client.connect();
+
+ client.setMarshaller(RMMarshaller.getInstance());
+
+ if ((false == oneWay) && (null == backPort))
+ client.setUnMarshaller(RMUnMarshaller.getInstance());
+
+ Map<String, Object> remotingInvocationContext =
rmRequest.getMetadata().getContext(REMOTING_INVOCATION_CONTEXT);
+ if (logger.isDebugEnabled())
+ logger.debug("Remoting metadata: " + remotingInvocationContext);
+
+ // debug the outgoing request message
+ MessageTrace.traceMessage("Outgoing RM Request Message",
rmRequest.getPayload());
+
+ RMMessage rmResponse = null;
+ if (oneWay && (null == backPort))
+ {
+ client.invokeOneway(rmRequest.getPayload(), remotingInvocationContext,
false);
+ }
+ else
+ {
+ Object retVal = client.invoke(rmRequest.getPayload(),
remotingInvocationContext);
+ if ((null != retVal) && (false == (retVal instanceof RMMessage)))
+ {
+ String msg = retVal.getClass().getName() + ": '" + retVal +
"'";
+ logger.warn(msg);
+ throw new RuntimeException(msg);
+ }
+ rmResponse = (RMMessage)retVal;
+ }
+
+ // Disconnect the remoting client
+ client.disconnect();
+
+ // trace the incomming response message
+ if (rmResponse != null)
+ MessageTrace.traceMessage("Incoming RM Response Message",
rmResponse.getPayload());
+
+ if (backPort != null) // TODO: backport support
+ {
+ if ((null != messageId) && (false ==
RMTransportHelper.isOneWayOperation(rmRequest)))
+ {
+ // register callbacks only for outbound messages with messageId
+ return new RMChannelResponse(callbackHandler, messageId);
+ }
+ }
+
+ return new RMChannelResponse(rmResponse);
+ }
+ catch (Throwable t)
+ {
+ return new RMChannelResponse(t);
+ }
+ }
+}
\ No newline at end of file
Property changes on:
stack/native/trunk/src/main/java/org/jboss/ws/extensions/wsrm/transport/RMChannelTask.java
___________________________________________________________________
Name: svn:keywords
+ Id Revision
Name: svn:eol-style
+ LF