svn commit: r523888 - in /activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf: CxfConsumer.java CxfEndpoint.java CxfExchange.java

[js...@apache.org]

Author: jstrachan
Date: Thu Mar 29 17:05:25 2007
New Revision: 523888

URL: http://svn.apache.org/viewvc?view=rev&rev=523888
Log:
added a CXF consumer

Added:
    activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java   (with props)
Modified:
    activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java
    activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfExchange.java

Added: activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java?view=auto&rev=523888
==============================================================================
--- activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java (added)
+++ activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java Thu Mar 29 17:05:25 2007
@@ -0,0 +1,65 @@
+/**
+ *
+ * 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.camel.component.cxf;
+
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.Processor;
+import org.apache.cxf.transport.local.LocalDestination;
+import org.apache.cxf.transport.MessageObserver;
+import org.apache.cxf.message.Message;
+
+/**
+ * A consumer of exchanges for a service in CXF
+ *
+ * @version $Revision$
+*/
+public class CxfConsumer extends DefaultConsumer<CxfExchange> {
+    private CxfEndpoint endpoint;
+    private LocalDestination destination;
+
+    public CxfConsumer(CxfEndpoint endpoint, Processor<CxfExchange> processor) {
+        super(endpoint, processor);
+        this.endpoint = endpoint;
+    }
+
+
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+
+        destination = (LocalDestination) endpoint.getLocalTransportFactory().getDestination(endpoint.getEndpointInfo());
+        destination.setMessageObserver(new MessageObserver() {
+            public void onMessage(Message message) {
+                incomingCxfMessage(message);
+            }
+        });
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        if (destination != null) {
+            destination.shutdown();
+        }
+        super.doStop();
+    }
+
+    protected void incomingCxfMessage(Message message) {
+        CxfExchange exchange = endpoint.createExchange(message);
+        getProcessor().onExchange(exchange);
+    }
+}

Propchange: activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
------------------------------------------------------------------------------
    svn:keywords = Date Author Id Revision HeadURL

Propchange: activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfConsumer.java
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Modified: activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java?view=diff&rev=523888&r1=523887&r2=523888
==============================================================================
--- activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java (original)
+++ activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfEndpoint.java Thu Mar 29 17:05:25 2007
@@ -20,11 +20,10 @@
 import org.apache.camel.Consumer;
 import org.apache.camel.Processor;
 import org.apache.camel.Producer;
-import org.apache.camel.impl.DefaultConsumer;
 import org.apache.camel.impl.DefaultEndpoint;
-import org.apache.cxf.endpoint.Server;
 import org.apache.cxf.transport.local.LocalTransportFactory;
 import org.apache.cxf.service.model.EndpointInfo;
+import org.apache.cxf.message.Message;
 
 /**
  * The endpoint in the service engine
@@ -48,12 +47,15 @@
     }
 
     public Consumer<CxfExchange> createConsumer(Processor<CxfExchange> processor) throws Exception {
-        return startService(new DefaultConsumer<CxfExchange>(this, processor) {
-        });
+        return startService(new CxfConsumer(this, processor));
     }
 
     public CxfExchange createExchange() {
         return new CxfExchange(getContext(), getBinding());
+    }
+
+    public CxfExchange createExchange(Message inMessage) {
+        return new CxfExchange(getContext(), getBinding(), inMessage);
     }
 
     public CxfBinding getBinding() {

Modified: activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfExchange.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfExchange.java?view=diff&rev=523888&r1=523887&r2=523888
==============================================================================
--- activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfExchange.java (original)
+++ activemq/camel/trunk/camel-cxf/src/main/java/org/apache/camel/component/cxf/CxfExchange.java Thu Mar 29 17:05:25 2007
@@ -48,6 +48,18 @@
         setFault(new CxfMessage(exchange.getInFaultMessage()));
     }
 
+    public CxfExchange(CamelContext context, CxfBinding binding, Message inMessage) {
+        super(context);
+        this.binding = binding;
+        this.exchange = inMessage.getExchange();
+
+        setIn(new CxfMessage(inMessage));
+        if (exchange != null) {
+            setOut(new CxfMessage(exchange.getOutMessage()));
+            setFault(new CxfMessage(exchange.getInFaultMessage()));
+        }
+    }
+
     @Override
     public CxfMessage getIn() {
         return (CxfMessage) super.getIn();