You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tubemq.apache.org by la...@apache.org on 2020/04/30 09:20:06 UTC
[incubator-tubemq] branch revert-67-master created (now 99b5975)
This is an automated email from the ASF dual-hosted git repository.
lamberliu pushed a change to branch revert-67-master
in repository https://gitbox.apache.org/repos/asf/incubator-tubemq.git.
at 99b5975 Revert "[TUBEMQ-85] There is NPE when creating PullConsumer with TubeSingleSessionFactory (#67)"
This branch includes the following new commits:
new 99b5975 Revert "[TUBEMQ-85] There is NPE when creating PullConsumer with TubeSingleSessionFactory (#67)"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[incubator-tubemq] 01/01: Revert "[TUBEMQ-85] There is NPE when
creating PullConsumer with TubeSingleSessionFactory (#67)"
Posted by la...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lamberliu pushed a commit to branch revert-67-master
in repository https://gitbox.apache.org/repos/asf/incubator-tubemq.git
commit 99b5975a79edccd9b97780595f358bb79dbaf3a9
Author: lamberliu <55...@users.noreply.github.com>
AuthorDate: Thu Apr 30 17:20:01 2020 +0800
Revert "[TUBEMQ-85] There is NPE when creating PullConsumer with TubeSingleSessionFactory (#67)"
This reverts commit 786656958628b27f48c20745eb9697c329ea72dc.
---
.../client/factory/TubeMultiSessionFactory.java | 15 +------------
.../client/factory/TubeSingleSessionFactory.java | 26 ----------------------
2 files changed, 1 insertion(+), 40 deletions(-)
diff --git a/tubemq-client/src/main/java/org/apache/tubemq/client/factory/TubeMultiSessionFactory.java b/tubemq-client/src/main/java/org/apache/tubemq/client/factory/TubeMultiSessionFactory.java
index 1cf91be..3dbb561 100644
--- a/tubemq-client/src/main/java/org/apache/tubemq/client/factory/TubeMultiSessionFactory.java
+++ b/tubemq-client/src/main/java/org/apache/tubemq/client/factory/TubeMultiSessionFactory.java
@@ -34,13 +34,12 @@ public class TubeMultiSessionFactory implements MessageSessionFactory {
private final NettyClientFactory clientFactory = new NettyClientFactory();
private final TubeBaseSessionFactory baseSessionFactory;
- private final AtomicBoolean isShutDown = new AtomicBoolean(true);
+ private final AtomicBoolean isShutDown = new AtomicBoolean(false);
public TubeMultiSessionFactory(final TubeClientConfig tubeClientConfig) throws TubeClientException {
RpcConfig config = TubeClientConfigUtils.getRpcConfigByClientConfig(tubeClientConfig, false);
clientFactory.configure(config);
baseSessionFactory = new TubeBaseSessionFactory(clientFactory, tubeClientConfig);
- isShutDown.set(false);
}
@Override
@@ -53,35 +52,23 @@ public class TubeMultiSessionFactory implements MessageSessionFactory {
@Override
public <T extends Shutdownable> void removeClient(final T client) {
- if (baseSessionFactory == null) {
- return;
- }
this.baseSessionFactory.removeClient(client);
}
@Override
public MessageProducer createProducer() throws TubeClientException {
- if (isShutDown.get()) {
- throw new TubeClientException("Please initialize the object first!");
- }
return this.baseSessionFactory.createProducer();
}
@Override
public PushMessageConsumer createPushConsumer(final ConsumerConfig consumerConfig)
throws TubeClientException {
- if (isShutDown.get()) {
- throw new TubeClientException("Please initialize the object first!");
- }
return this.baseSessionFactory.createPushConsumer(consumerConfig);
}
@Override
public PullMessageConsumer createPullConsumer(ConsumerConfig consumerConfig)
throws TubeClientException {
- if (isShutDown.get()) {
- throw new TubeClientException("Please initialize the object first!");
- }
return this.baseSessionFactory.createPullConsumer(consumerConfig);
}
diff --git a/tubemq-client/src/main/java/org/apache/tubemq/client/factory/TubeSingleSessionFactory.java b/tubemq-client/src/main/java/org/apache/tubemq/client/factory/TubeSingleSessionFactory.java
index 1532645..a574bf4 100644
--- a/tubemq-client/src/main/java/org/apache/tubemq/client/factory/TubeSingleSessionFactory.java
+++ b/tubemq-client/src/main/java/org/apache/tubemq/client/factory/TubeSingleSessionFactory.java
@@ -17,7 +17,6 @@
package org.apache.tubemq.client.factory;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.tubemq.client.config.ConsumerConfig;
import org.apache.tubemq.client.config.TubeClientConfig;
@@ -34,7 +33,6 @@ import org.apache.tubemq.corerpc.netty.NettyClientFactory;
public class TubeSingleSessionFactory implements MessageSessionFactory {
private static final NettyClientFactory clientFactory = new NettyClientFactory();
- private static final AtomicBoolean isShutDown = new AtomicBoolean(true);
private static final AtomicLong referenceCounter = new AtomicLong(0);
private static TubeBaseSessionFactory baseSessionFactory;
@@ -44,61 +42,37 @@ public class TubeSingleSessionFactory implements MessageSessionFactory {
RpcConfig config = TubeClientConfigUtils.getRpcConfigByClientConfig(tubeClientConfig, true);
clientFactory.configure(config);
baseSessionFactory = new TubeBaseSessionFactory(clientFactory, tubeClientConfig);
- isShutDown.set(false);
- }
- while (isShutDown.get()) {
- try {
- Thread.sleep(50);
- } catch (Throwable e) {
- break;
- }
}
}
@Override
public void shutdown() throws TubeClientException {
- if (isShutDown.get()) {
- throw new TubeClientException("Please initialize the object first!");
- }
if (referenceCounter.decrementAndGet() > 0) {
return;
}
baseSessionFactory.shutdown();
clientFactory.shutdown();
- isShutDown.set(true);
}
@Override
public <T extends Shutdownable> void removeClient(final T client) {
- if (baseSessionFactory == null) {
- return;
- }
baseSessionFactory.removeClient(client);
}
@Override
public MessageProducer createProducer() throws TubeClientException {
- if (isShutDown.get()) {
- throw new TubeClientException("Please initialize the object first!");
- }
return baseSessionFactory.createProducer();
}
@Override
public PushMessageConsumer createPushConsumer(ConsumerConfig consumerConfig)
throws TubeClientException {
- if (isShutDown.get()) {
- throw new TubeClientException("Please initialize the object first!");
- }
return baseSessionFactory.createPushConsumer(consumerConfig);
}
@Override
public PullMessageConsumer createPullConsumer(ConsumerConfig consumerConfig)
throws TubeClientException {
- if (isShutDown.get()) {
- throw new TubeClientException("Please initialize the object first!");
- }
return baseSessionFactory.createPullConsumer(consumerConfig);
}