You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2015/08/19 08:58:47 UTC
[1/2] camel git commit: CAMEL-9087: camel-pgevent payload always
null. Thanks to Marek Schmidt for the patch.
Repository: camel
Updated Branches:
refs/heads/camel-2.15.x d4791de1e -> 38e0233bc
refs/heads/master dedb59d82 -> bedfdeb8f
CAMEL-9087: camel-pgevent payload always null. Thanks to Marek Schmidt for the patch.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/bedfdeb8
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/bedfdeb8
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/bedfdeb8
Branch: refs/heads/master
Commit: bedfdeb8f78e910cd48a503e142d784942ba3770
Parents: dedb59d
Author: Claus Ibsen <da...@apache.org>
Authored: Wed Aug 19 09:06:22 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Aug 19 09:06:22 2015 +0200
----------------------------------------------------------------------
.../component/pgevent/PgEventConsumer.java | 24 ++++++++++++--------
.../component/pgevent/PgEventEndpoint.java | 1 +
.../component/pgevent/PgEventProducer.java | 17 ++++++--------
3 files changed, 23 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/bedfdeb8/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
index ed93807..a0a5f2c 100644
--- a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
+++ b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
@@ -49,22 +49,28 @@ public class PgEventConsumer extends DefaultConsumer implements PGNotificationLi
@Override
public void notification(int processId, String channel, String payload) {
- Exchange outOnly = endpoint.createExchange();
- Message msg = outOnly.getOut();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Notification processId: {}, channel: {}, payload: {}", new Object[]{processId, channel, payload});
+ }
+
+ Exchange exchange = endpoint.createExchange();
+ Message msg = exchange.getIn();
msg.setHeader("channel", channel);
msg.setBody(payload);
- outOnly.setOut(msg);
+
try {
- getProcessor().process(outOnly);
+ getProcessor().process(exchange);
} catch (Exception ex) {
- LOG.error("Unable to process incoming notification from PostgreSQL: processId='" + processId + "', channel='" + channel + "', payload='" + payload + "'", ex);
+ String cause = "Unable to process incoming notification from PostgreSQL: processId='" + processId + "', channel='" + channel + "', payload='" + payload + "'";
+ getExceptionHandler().handleException(cause, ex);
}
}
@Override
- protected void doShutdown() throws Exception {
- super.doShutdown();
- dbConnection.removeNotificationListener(endpoint.getChannel());
- dbConnection.close();
+ protected void doStop() throws Exception {
+ if (dbConnection != null) {
+ dbConnection.removeNotificationListener(endpoint.getChannel());
+ dbConnection.close();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/bedfdeb8/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java
index 774b3f4..ca3ff39 100644
--- a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java
+++ b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java
@@ -161,6 +161,7 @@ public class PgEventEndpoint extends DefaultEndpoint {
public Consumer createConsumer(Processor processor) throws Exception {
validateInputs();
PgEventConsumer consumer = new PgEventConsumer(this, processor);
+ configureConsumer(consumer);
return consumer;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/bedfdeb8/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
index 0b29619..2f45dde 100644
--- a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
+++ b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
@@ -19,21 +19,16 @@ package org.apache.camel.component.pgevent;
import java.sql.SQLException;
import com.impossibl.postgres.api.jdbc.PGConnection;
-import com.impossibl.postgres.api.jdbc.PGNotificationListener;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.impl.DefaultAsyncProducer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* The PgEvent producer.
*/
public class PgEventProducer extends DefaultAsyncProducer {
- private static final Logger LOG = LoggerFactory.getLogger(PgEventProducer.class);
private final PgEventEndpoint endpoint;
private PGConnection dbConnection;
- private PGNotificationListener listener;
public PgEventProducer(PgEventEndpoint endpoint) throws Exception {
super(endpoint);
@@ -53,7 +48,7 @@ public class PgEventProducer extends DefaultAsyncProducer {
}
try {
- dbConnection.createStatement().execute("NOTIFY " + endpoint.getChannel() + ", '" + exchange.getOut().getBody(String.class) + "'");
+ dbConnection.createStatement().execute("NOTIFY " + endpoint.getChannel() + ", '" + exchange.getIn().getBody(String.class) + "'");
} catch (SQLException e) {
exchange.setException(e);
}
@@ -68,9 +63,11 @@ public class PgEventProducer extends DefaultAsyncProducer {
}
@Override
- protected void doShutdown() throws Exception {
- super.doShutdown();
- dbConnection.close();
+ protected void doStop() throws Exception {
+ super.doStop();
+ if (dbConnection != null) {
+ dbConnection.close();
+ }
}
-}
\ No newline at end of file
+}
[2/2] camel git commit: CAMEL-9087: camel-pgevent payload always
null. Thanks to Marek Schmidt for the patch.
Posted by da...@apache.org.
CAMEL-9087: camel-pgevent payload always null. Thanks to Marek Schmidt for the patch.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/38e0233b
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/38e0233b
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/38e0233b
Branch: refs/heads/camel-2.15.x
Commit: 38e0233bce0328a5089b058c65b1e82fabf1bc7d
Parents: d4791de
Author: Claus Ibsen <da...@apache.org>
Authored: Wed Aug 19 09:06:22 2015 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Wed Aug 19 09:06:55 2015 +0200
----------------------------------------------------------------------
.../component/pgevent/PgEventConsumer.java | 24 ++++++++++++--------
.../component/pgevent/PgEventEndpoint.java | 1 +
.../component/pgevent/PgEventProducer.java | 17 ++++++--------
3 files changed, 23 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/38e0233b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
index ed93807..a0a5f2c 100644
--- a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
+++ b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventConsumer.java
@@ -49,22 +49,28 @@ public class PgEventConsumer extends DefaultConsumer implements PGNotificationLi
@Override
public void notification(int processId, String channel, String payload) {
- Exchange outOnly = endpoint.createExchange();
- Message msg = outOnly.getOut();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Notification processId: {}, channel: {}, payload: {}", new Object[]{processId, channel, payload});
+ }
+
+ Exchange exchange = endpoint.createExchange();
+ Message msg = exchange.getIn();
msg.setHeader("channel", channel);
msg.setBody(payload);
- outOnly.setOut(msg);
+
try {
- getProcessor().process(outOnly);
+ getProcessor().process(exchange);
} catch (Exception ex) {
- LOG.error("Unable to process incoming notification from PostgreSQL: processId='" + processId + "', channel='" + channel + "', payload='" + payload + "'", ex);
+ String cause = "Unable to process incoming notification from PostgreSQL: processId='" + processId + "', channel='" + channel + "', payload='" + payload + "'";
+ getExceptionHandler().handleException(cause, ex);
}
}
@Override
- protected void doShutdown() throws Exception {
- super.doShutdown();
- dbConnection.removeNotificationListener(endpoint.getChannel());
- dbConnection.close();
+ protected void doStop() throws Exception {
+ if (dbConnection != null) {
+ dbConnection.removeNotificationListener(endpoint.getChannel());
+ dbConnection.close();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/38e0233b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java
index 197d48e..b1d2610 100644
--- a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java
+++ b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventEndpoint.java
@@ -162,6 +162,7 @@ public class PgEventEndpoint extends DefaultEndpoint {
public Consumer createConsumer(Processor processor) throws Exception {
validateInputs();
PgEventConsumer consumer = new PgEventConsumer(this, processor);
+ configureConsumer(consumer);
return consumer;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/38e0233b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
index 0b29619..2f45dde 100644
--- a/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
+++ b/components/camel-pgevent/src/main/java/org/apache/camel/component/pgevent/PgEventProducer.java
@@ -19,21 +19,16 @@ package org.apache.camel.component.pgevent;
import java.sql.SQLException;
import com.impossibl.postgres.api.jdbc.PGConnection;
-import com.impossibl.postgres.api.jdbc.PGNotificationListener;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
import org.apache.camel.impl.DefaultAsyncProducer;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* The PgEvent producer.
*/
public class PgEventProducer extends DefaultAsyncProducer {
- private static final Logger LOG = LoggerFactory.getLogger(PgEventProducer.class);
private final PgEventEndpoint endpoint;
private PGConnection dbConnection;
- private PGNotificationListener listener;
public PgEventProducer(PgEventEndpoint endpoint) throws Exception {
super(endpoint);
@@ -53,7 +48,7 @@ public class PgEventProducer extends DefaultAsyncProducer {
}
try {
- dbConnection.createStatement().execute("NOTIFY " + endpoint.getChannel() + ", '" + exchange.getOut().getBody(String.class) + "'");
+ dbConnection.createStatement().execute("NOTIFY " + endpoint.getChannel() + ", '" + exchange.getIn().getBody(String.class) + "'");
} catch (SQLException e) {
exchange.setException(e);
}
@@ -68,9 +63,11 @@ public class PgEventProducer extends DefaultAsyncProducer {
}
@Override
- protected void doShutdown() throws Exception {
- super.doShutdown();
- dbConnection.close();
+ protected void doStop() throws Exception {
+ super.doStop();
+ if (dbConnection != null) {
+ dbConnection.close();
+ }
}
-}
\ No newline at end of file
+}