Skip to content

Commit 21ad2a4

Browse files
sbordetsbordet
committed
Fixes COMETD-427 (Improve BayeuxServer.createIfAbsent()).
Introduced method: MarkedReference<ServerChannel> BayeuxServer.createChannelIfAbsent(String, Initializer...) that returns the channel and a flag telling whether the channel was created or it already existed. Deprecated BayeuxServer.createIfAbsent(String, Initializer...) and updated all usages to the new method.
1 parent 2bfee7a commit 21ad2a4

35 files changed

Lines changed: 173 additions & 244 deletions

File tree

‎cometd-demo/src/main/java/org/webtide/demo/auction/AuctionChatService.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -58,14 +58,14 @@ public AuctionChatService(ServletContext context)
5858
throw new RuntimeException("Missing " + Seti.SETI_ATTRIBUTE + " from " + ServletContext.class.getSimpleName() + "; " +
5959
"is " + SetiServlet.class.getSimpleName() + " declared in web.xml ?");
6060

61-
getBayeux().createIfAbsent("/auction/chat/**", new ConfigurableServerChannel.Initializer()
61+
getBayeux().createChannelIfAbsent("/auction/chat/**", new ConfigurableServerChannel.Initializer()
6262
{
6363
public void configureChannel(ConfigurableServerChannel channel)
6464
{
6565
channel.addAuthorizer(GrantAuthorizer.GRANT_ALL);
6666
}
6767
});
68-
getBayeux().createIfAbsent("/service/auction/chat", new ConfigurableServerChannel.Initializer()
68+
getBayeux().createChannelIfAbsent("/service/auction/chat", new ConfigurableServerChannel.Initializer()
6969
{
7070
public void configureChannel(ConfigurableServerChannel channel)
7171
{

‎cometd-demo/src/main/java/org/webtide/demo/auction/AuctionService.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -69,14 +69,14 @@ public AuctionService(ServletContext context)
6969
getBayeux().addListener(this);
7070
setSeeOwnPublishes(false);
7171

72-
getBayeux().createIfAbsent("/service" + AUCTION_ROOT + "*", new ConfigurableServerChannel.Initializer()
72+
getBayeux().createChannelIfAbsent("/service" + AUCTION_ROOT + "*", new ConfigurableServerChannel.Initializer()
7373
{
7474
public void configureChannel(ConfigurableServerChannel channel)
7575
{
7676
channel.addAuthorizer(GrantAuthorizer.GRANT_ALL);
7777
}
7878
});
79-
getBayeux().createIfAbsent(AUCTION_ROOT + "*", new ConfigurableServerChannel.Initializer()
79+
getBayeux().createChannelIfAbsent(AUCTION_ROOT + "*", new ConfigurableServerChannel.Initializer()
8080
{
8181
public void configureChannel(ConfigurableServerChannel channel)
8282
{

cometd-java/cometd-java-common/src/main/java/org/cometd/common/MarkedReference.java renamed to cometd-java/bayeux-api/src/main/java/org/cometd/bayeux/MarkedReference.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,12 +14,13 @@
1414
* limitations under the License.
1515
*/
1616

17-
package org.cometd.common;
17+
package org.cometd.bayeux;
1818

1919
import java.util.concurrent.atomic.AtomicMarkableReference;
2020

2121
/**
2222
* Immutable, non-volatile, non-atomic version of {@link AtomicMarkableReference}.
23+
*
2324
* @param <T> the reference type
2425
*/
2526
public class MarkedReference<T>

‎cometd-java/bayeux-api/src/main/java/org/cometd/bayeux/server/BayeuxServer.java‎

Lines changed: 19 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import java.util.List;
2020

2121
import org.cometd.bayeux.Bayeux;
22+
import org.cometd.bayeux.MarkedReference;
2223
import org.cometd.bayeux.Transport;
2324
import org.cometd.bayeux.client.ClientSession;
2425
import org.cometd.bayeux.client.ClientSessionChannel;
@@ -89,7 +90,7 @@ public interface BayeuxServer extends Bayeux
8990
* @param channelId the channel identifier
9091
* @return a {@link ServerChannel} with the given {@code channelId},
9192
* or null if no such channel exists
92-
* @see #createIfAbsent(String, org.cometd.bayeux.server.ConfigurableServerChannel.Initializer...)
93+
* @see #createChannelIfAbsent(String, ConfigurableServerChannel.Initializer...)
9394
*/
9495
ServerChannel getChannel(String channelId);
9596

@@ -99,20 +100,30 @@ public interface BayeuxServer extends Bayeux
99100
List<ServerChannel> getChannels();
100101

101102
/**
102-
* <p>Creates a {@link ServerChannel} and initializes it atomically.</p>
103+
* @deprecated use {@link #createChannelIfAbsent(String, ConfigurableServerChannel.Initializer...)}
104+
*/
105+
@Deprecated
106+
boolean createIfAbsent(String channelId, ConfigurableServerChannel.Initializer... initializers);
107+
108+
/**
109+
* <p>Creates a {@link ServerChannel} and initializes it atomically if the
110+
* channel does not exist, or returns it if it already exists.</p>
103111
* <p>This method can be used instead of adding a {@link ChannelListener}
104-
* to atomically initialize a channel. The initializer will be called before
105-
* any other thread can access the new channel instance.</p>
106-
* <p>The createIfAbsent method should be used when a channel needs to be
112+
* to atomically initialize a channel. The {@code initializers} will be
113+
* called before any other thread can access the new channel instance.</p>
114+
* <p>Method {@link #createChannelIfAbsent(String, ConfigurableServerChannel.Initializer...)}
115+
* should be used when a channel needs to be
107116
* initialized (e.g. by adding listeners) before any publish or subscribes
108117
* can occur on the channel, or before any other thread may concurrently
109118
* create the same channel.</p>
110119
*
111-
* @param channelId the channel identifier
120+
* @param channelName the channel name
112121
* @param initializers the initializers invoked to configure the channel
113-
* @return true if the channel was initialized, false otherwise
122+
* @return a {@link MarkedReference} whose reference is the channel, and
123+
* the mark signals whether the channel has been created because it
124+
* did not exist before.
114125
*/
115-
boolean createIfAbsent(String channelId, ConfigurableServerChannel.Initializer... initializers);
126+
MarkedReference<ServerChannel> createChannelIfAbsent(String channelName, ConfigurableServerChannel.Initializer... initializers);
116127

117128
/**
118129
* @param clientId the {@link ServerSession} identifier

‎cometd-java/bayeux-api/src/main/java/org/cometd/bayeux/server/ConfigurableServerChannel.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
* <p>A {@link ConfigurableServerChannel} offers an API that can be used to
2626
* configure {@link ServerChannel}s at creation time.</p>
2727
* <p>{@link ServerChannel}s may be created concurrently via
28-
* {@link BayeuxServer#createIfAbsent(String, Initializer...)} and it is
28+
* {@link BayeuxServer#createChannelIfAbsent(String, Initializer...)} and it is
2929
* important that the creation of a channel is atomic so that its
3030
* configuration is executed only once, and so that it is guaranteed that
3131
* it happens before any message can be published or received by the channel.</p>

‎cometd-java/bayeux-api/src/main/java/org/cometd/bayeux/server/SecurityPolicy.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,7 @@ public interface SecurityPolicy
6262
* {@link ClientSessionChannel#subscribe(ClientSessionChannel.MessageListener)} or
6363
* {@link ClientSessionChannel#publish(Object)} are therefore subject to this check.
6464
* <p />
65-
* Direct calls to {@link BayeuxServer#createIfAbsent(String, ConfigurableServerChannel.Initializer...)}
65+
* Direct calls to {@link BayeuxServer#createChannelIfAbsent(String, ConfigurableServerChannel.Initializer...)}
6666
* are not subject to this check.
6767
*
6868
* @param server the {@link BayeuxServer} object

‎cometd-java/cometd-java-annotations/src/main/java/org/cometd/annotation/ServerAnnotationProcessor.java‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import javax.annotation.PreDestroy;
3131
import javax.inject.Inject;
3232

33+
import org.cometd.bayeux.MarkedReference;
3334
import org.cometd.bayeux.Message;
3435
import org.cometd.bayeux.client.ClientSessionChannel;
3536
import org.cometd.bayeux.server.BayeuxServer;
@@ -168,9 +169,9 @@ public void configureChannel(ConfigurableServerChannel channel)
168169
}
169170
};
170171

171-
boolean initialized = bayeuxServer.createIfAbsent(channel, init);
172+
MarkedReference<ServerChannel> initializedChannel = bayeuxServer.createChannelIfAbsent(channel, init);
172173

173-
if (initialized)
174+
if (initializedChannel.isMarked())
174175
{
175176
logger.debug("Channel {} already initialized. Not called method {} on bean {}", channel, method, bean);
176177
}
@@ -179,10 +180,12 @@ public void configureChannel(ConfigurableServerChannel channel)
179180
if (configure.configureIfExists())
180181
{
181182
logger.debug("Configure channel {} with method {} on bean {}", channel, method, bean);
182-
init.configureChannel(bayeuxServer.getChannel(channel));
183+
init.configureChannel(initializedChannel.getReference());
183184
}
184185
else if (configure.errorIfExists())
186+
{
185187
throw new IllegalStateException("Channel already configured: " + channel);
188+
}
186189
}
187190
}
188191
}
@@ -388,9 +391,9 @@ private boolean processListener(Object bean, LocalSession localSession)
388391
String[] channels = listener.value();
389392
for (String channel : channels)
390393
{
391-
bayeuxServer.createIfAbsent(channel);
394+
MarkedReference<ServerChannel> initializedChannel = bayeuxServer.createChannelIfAbsent(channel);
392395
ListenerCallback listenerCallback = new ListenerCallback(localSession, bean, method, channel, listener.receiveOwnPublishes());
393-
bayeuxServer.getChannel(channel).addListener(listenerCallback);
396+
initializedChannel.getReference().addListener(listenerCallback);
394397

395398
List<ListenerCallback> callbacks = listeners.get(bean);
396399
if (callbacks == null)

‎cometd-java/cometd-java-annotations/src/test/java/org/cometd/annotation/ServerAnnotationProcessorTest.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import javax.annotation.PreDestroy;
3030
import javax.inject.Inject;
3131

32+
import org.cometd.bayeux.MarkedReference;
3233
import org.cometd.bayeux.Message;
3334
import org.cometd.bayeux.server.BayeuxServer;
3435
import org.cometd.bayeux.server.ConfigurableServerChannel;
@@ -328,13 +329,13 @@ public void foo(ServerSession remote, ServerMessage.Mutable message)
328329
int count = counter.incrementAndGet();
329330

330331
String channelName = "/foo/own";
331-
bayeuxServer.createIfAbsent(channelName);
332+
MarkedReference<ServerChannel> channel = bayeuxServer.createChannelIfAbsent(channelName);
332333

333334
// This callback should be called only once, triggered by the client's publish
334335
// However if the Listener.receiveOwnPublishes attribute is not taken in account
335336
// this callback is called again, and we want to test that this does not happen.
336337
if (count == 1)
337-
bayeuxServer.getChannel(channelName).publish(serverSession, new HashMap(), null);
338+
channel.getReference().publish(serverSession, new HashMap(), null);
338339
}
339340
}
340341

@@ -376,9 +377,9 @@ public void foo(ServerSession remote, ServerMessage.Mutable message)
376377
{
377378
counter.incrementAndGet();
378379
String channelName = "/foo/own";
379-
bayeuxServer.createIfAbsent(channelName);
380+
MarkedReference<ServerChannel> channel = bayeuxServer.createChannelIfAbsent(channelName);
380381
if (!channelName.equals(message.getChannel()))
381-
bayeuxServer.getChannel(channelName).publish(serverSession, new HashMap(), null);
382+
channel.getReference().publish(serverSession, new HashMap(), null);
382383
}
383384
}
384385

‎cometd-java/cometd-java-client/src/test/java/org/cometd/client/BayeuxClientExtensionTest.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424

2525
import org.cometd.bayeux.Channel;
2626
import org.cometd.bayeux.ChannelId;
27+
import org.cometd.bayeux.MarkedReference;
2728
import org.cometd.bayeux.Message;
2829
import org.cometd.bayeux.client.ClientSession;
2930
import org.cometd.bayeux.client.ClientSessionChannel;
@@ -215,8 +216,8 @@ public void testReturningFalseOnSend() throws Exception
215216
{
216217
String channelName = "/test";
217218
final CountDownLatch latch = new CountDownLatch(1);
218-
bayeux.createIfAbsent(channelName);
219-
bayeux.getChannel(channelName).addListener(new ServerChannel.MessageListener()
219+
MarkedReference<ServerChannel> channel = bayeux.createChannelIfAbsent(channelName);
220+
channel.getReference().addListener(new ServerChannel.MessageListener()
220221
{
221222
public boolean onMessage(ServerSession from, ServerChannel channel, ServerMessage.Mutable message)
222223
{

‎cometd-java/cometd-java-client/src/test/java/org/cometd/client/BayeuxClientTest.java‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939

4040
import org.cometd.bayeux.Channel;
4141
import org.cometd.bayeux.ChannelId;
42+
import org.cometd.bayeux.MarkedReference;
4243
import org.cometd.bayeux.Message;
4344
import org.cometd.bayeux.client.ClientSessionChannel;
4445
import org.cometd.bayeux.server.BayeuxServer;
@@ -542,8 +543,8 @@ public void testPublish() throws Exception
542543
final BlockingArrayQueue<String> results = new BlockingArrayQueue<String>();
543544

544545
String channelName = "/chat/msg";
545-
bayeux.createIfAbsent(channelName);
546-
bayeux.getChannel(channelName).addListener(new ServerChannel.MessageListener()
546+
MarkedReference<ServerChannel> channel = bayeux.createChannelIfAbsent(channelName);
547+
channel.getReference().addListener(new ServerChannel.MessageListener()
547548
{
548549
public boolean onMessage(ServerSession from, ServerChannel channel, Mutable message)
549550
{
@@ -575,8 +576,8 @@ public void testWaitFor() throws Exception
575576
final BlockingArrayQueue<String> results = new BlockingArrayQueue<String>();
576577

577578
String channelName = "/chat/msg";
578-
bayeux.createIfAbsent(channelName);
579-
bayeux.getChannel(channelName).addListener(new ServerChannel.MessageListener()
579+
MarkedReference<ServerChannel> channel = bayeux.createChannelIfAbsent(channelName);
580+
channel.getReference().addListener(new ServerChannel.MessageListener()
580581
{
581582
public boolean onMessage(ServerSession from, ServerChannel channel, Mutable message)
582583
{

0 commit comments

Comments
 (0)