1 /*
2 * ModeShape (http://www.modeshape.org)
3 * See the COPYRIGHT.txt file distributed with this work for information
4 * regarding copyright ownership. Some portions may be licensed
5 * to Red Hat, Inc. under one or more contributor license agreements.
6 * See the AUTHORS.txt file in the distribution for a full listing of
7 * individual contributors.
8 *
9 * ModeShape is free software. Unless otherwise indicated, all code in ModeShape
10 * is licensed to you under the terms of the GNU Lesser General Public License as
11 * published by the Free Software Foundation; either version 2.1 of
12 * the License, or (at your option) any later version.
13 *
14 * ModeShape is distributed in the hope that it will be useful,
15 * but WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17 * Lesser General Public License for more details.
18 *
19 * You should have received a copy of the GNU Lesser General Public
20 * License along with this software; if not, write to the Free
21 * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
22 * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
23 */
24 package org.modeshape.clustering;
25
26 import java.io.ByteArrayInputStream;
27 import java.io.IOException;
28 import java.io.InputStream;
29 import java.util.concurrent.atomic.AtomicBoolean;
30 import org.jgroups.Address;
31 import org.jgroups.Channel;
32 import org.jgroups.ChannelClosedException;
33 import org.jgroups.ChannelException;
34 import org.jgroups.ChannelListener;
35 import org.jgroups.ChannelNotConnectedException;
36 import org.jgroups.JChannel;
37 import org.jgroups.Message;
38 import org.jgroups.View;
39 import org.jgroups.conf.ProtocolStackConfigurator;
40 import org.jgroups.conf.XmlConfigurator;
41 import org.jgroups.util.Util;
42 import org.modeshape.common.SystemFailureException;
43 import org.modeshape.common.util.CheckArg;
44 import org.modeshape.common.util.Logger;
45 import org.modeshape.graph.observe.ChangeObservers;
46 import org.modeshape.graph.observe.Changes;
47 import org.modeshape.graph.observe.ObservationBus;
48 import org.modeshape.graph.observe.Observer;
49
50 /**
51 * An implementation of a cluster-aware {@link ObservationBus}.
52 */
53 public class ClusteredObservationBus implements ObservationBus {
54
55 protected static final Logger LOGGER = Logger.getLogger(ClusteredObservationBus.class);
56
57 /**
58 * The list of {@link Observer} instances that should receive events from the bus. These Observer objects are all local.
59 */
60 protected final ChangeObservers observers = new ChangeObservers();
61
62 /**
63 * The listener for channel changes.
64 */
65 private final Listener listener = new Listener();
66
67 /**
68 * The component that will receive the JGroups messages and broadcast them to this bus' observers.
69 */
70 private final Receiver receiver = new Receiver();
71
72 /**
73 * Flag that dictates whether this bus has connected to the cluster.
74 */
75 protected final AtomicBoolean isOpen = new AtomicBoolean(false);
76
77 /**
78 * Flag that dictates whether there are multiple participants in the cluster; if not, then the changes are propogated only to
79 * the local observers.
80 */
81 protected final AtomicBoolean multipleAddressesInCluster = new AtomicBoolean(false);
82
83 /**
84 * The configuration for JGroups
85 */
86 private String configuration;
87
88 /**
89 * The name of the JGroups cluster.
90 */
91 private String clusterName;
92
93 /**
94 * The JGroups channel to which all {@link #notify(Changes) change notifications} will be sent and from which all changes will
95 * be received and sent to the observers.
96 * <p>
97 * It is important that the order of the {@link Changes} instances are maintained across the cluster, and JGroups will do this
98 * for us as long as we push all local changes into the channel and receive all local/remote changes from the channel.
99 * </p>
100 */
101 private JChannel channel;
102
103 /**
104 * Get the configuration for JGroups. This configuration may be a string that refers to a resource on the classpath, the path
105 * of the local configuration, the URL to the configuration file, or the configuration specified using the newer-style XML
106 * form or older-style string form.
107 *
108 * @return the location of the JGroups configuration, or null if no configuration has been defined
109 */
110 public String getConfiguration() {
111 return configuration;
112 }
113
114 /**
115 * Set the JGroups configuration, which may be a string that refers to a resource on the classpath, the path of the local
116 * configuration, the URL to the configuration file, or the configuration specified using the newer-style XML form or
117 * older-style string form.
118 *
119 * @param configuration the relative path to a classpath resource, path to a local file, URL to the configuration file, or the
120 * configuration specified using the newer-style XML form or older-style string form.
121 * @throws IllegalStateException if this method is called after this bus has been {@link #start() started} but before it has
122 * been {@link #shutdown() shutdown}
123 */
124 public void setConfiguration( String configuration ) {
125 if (channel != null) {
126 String name = this.clusterName;
127 throw new IllegalStateException(ClusteringI18n.clusteringChannelIsRunningAndCannotBeChangedUnlessShutdown.text(name));
128 }
129 this.configuration = configuration;
130 }
131
132 /**
133 * Get the name of the JGroups cluster.
134 *
135 * @return the cluster name, or null if the cluster name was not yet defined
136 * @see JChannel#connect(String)
137 */
138 public String getClusterName() {
139 return clusterName;
140 }
141
142 /**
143 * Set the name of the JGroups cluster. This must be called with a non-null cluster name before {@link #start()} is called.
144 *
145 * @param clusterName Sets clusterName to the specified value.
146 * @throws IllegalStateException if this method is called after this bus has been {@link #start() started} but before it has
147 * been {@link #shutdown() shutdown}
148 * @see JChannel#connect(String)
149 */
150 public void setClusterName( String clusterName ) {
151 CheckArg.isNotNull(clusterName, "clusterName");
152 if (channel != null) {
153 String name = this.clusterName;
154 throw new IllegalStateException(ClusteringI18n.clusteringChannelIsRunningAndCannotBeChangedUnlessShutdown.text(name));
155 }
156 this.clusterName = clusterName;
157 }
158
159 /**
160 * {@inheritDoc}
161 *
162 * @see org.modeshape.graph.observe.ObservationBus#start()
163 */
164 @Override
165 public synchronized void start() {
166 if (clusterName == null) {
167 throw new IllegalStateException(ClusteringI18n.clusterNameRequired.text());
168 }
169 if (channel != null) {
170 // Disconnect from any previous channel ...
171 channel.removeChannelListener(listener);
172 channel.setReceiver(null);
173 }
174 try {
175 // Create the new channel by calling the delegate method ...
176 channel = newChannel(configuration);
177 assert channel != null;
178 // Add a listener through which we'll know what's going on within the cluster ...
179 channel.addChannelListener(listener);
180
181 // Set the receiver through which we'll receive all of the changes ...
182 channel.setReceiver(receiver);
183
184 // Now connect to the cluster ...
185 channel.connect(clusterName);
186 } catch (ChannelException e) {
187 throw new IllegalStateException(ClusteringI18n.errorWhileStartingJGroups.text(configuration), e);
188 }
189 }
190
191 /**
192 * A method that is used to instantiate the {@link JChannel} object with the supplied configuration. Subclasses can override
193 * this method to specialize this behavior.
194 *
195 * @param configuration the configuration; may be null if the default configuration should be used
196 * @return the JChannel instance; never null
197 * @throws ChannelException if there is a problem creating the new channel object
198 */
199 protected JChannel newChannel( String configuration ) throws ChannelException {
200 if (configuration == null) {
201 return new JChannel();
202 }
203 // Try the XML configuration first ...
204 ProtocolStackConfigurator configurator = null;
205 InputStream stream = new ByteArrayInputStream(configuration.getBytes());
206 try {
207 configurator = XmlConfigurator.getInstance(stream);
208 } catch (IOException e) {
209 // ignore, since the configuration may be of another form ...
210 } finally {
211 try {
212 stream.close();
213 } catch (IOException e) {
214 // ignore this
215 }
216 }
217 if (configurator != null) {
218 return new JChannel(configurator);
219 }
220 // Otherwise, just try the regular configuration ...
221 return new JChannel(configuration);
222 }
223
224 /**
225 * {@inheritDoc}
226 *
227 * @see org.modeshape.graph.observe.Observer#notify(org.modeshape.graph.observe.Changes)
228 */
229 @Override
230 public void notify( Changes changes ) {
231 if (changes == null) return; // do nothing
232 if (!isOpen.get()) {
233 // The channel is not open ...
234 return;
235 }
236 if (!multipleAddressesInCluster.get()) {
237 // We are in clustered mode, but there is only one participant in the cluster (us).
238 // So short-circuit the cluster and just notify the local observers ...
239 if (!observers.isEmpty()) {
240 observers.broadcast(changes);
241 if (LOGGER.isTraceEnabled()) {
242 LOGGER.trace("Received on cluster '{0}' {1} changes in source '{2}' made by {3} from process '{4}' at {5}",
243 getClusterName(),
244 changes.getChangeRequests().size(),
245 changes.getSourceName(),
246 changes.getUserName(),
247 changes.getProcessId(),
248 changes.getTimestamp());
249 }
250 }
251 return;
252 }
253
254 // There are multiple participants in the cluster, so send all changes out to JGroups,
255 // letting JGroups do the ordering of messages...
256 try {
257 if (LOGGER.isTraceEnabled()) {
258 LOGGER.trace("Sending to cluster '{0}' {1} changes in source '{2}' made by {3} from process '{4}' at {5}",
259 clusterName,
260 changes.getChangeRequests().size(),
261 changes.getSourceName(),
262 changes.getUserName(),
263 changes.getProcessId(),
264 changes.getTimestamp());
265 }
266 byte[] data = serialize(changes);
267 Message message = new Message(null, null, data);
268 channel.send(message);
269 } catch (ChannelClosedException e) {
270 LOGGER.warn(ClusteringI18n.unableToNotifyChangesBecauseClusterChannelHasClosed,
271 clusterName,
272 changes.getChangeRequests().size(),
273 changes.getSourceName(),
274 changes.getUserName(),
275 changes.getProcessId(),
276 changes.getTimestamp());
277 } catch (ChannelNotConnectedException e) {
278 LOGGER.warn(ClusteringI18n.unableToNotifyChangesBecauseClusterChannelIsNotConnected,
279 clusterName,
280 changes.getChangeRequests().size(),
281 changes.getSourceName(),
282 changes.getUserName(),
283 changes.getProcessId(),
284 changes.getTimestamp());
285 } catch (Exception e) {
286 // Something went wrong here (this should not happen) ...
287 String msg = ClusteringI18n.errorSerializingChanges.text(clusterName,
288 changes.getChangeRequests().size(),
289 changes.getSourceName(),
290 changes.getUserName(),
291 changes.getProcessId(),
292 changes.getTimestamp(),
293 changes);
294 throw new SystemFailureException(msg, e);
295 }
296 }
297
298 /**
299 * {@inheritDoc}
300 *
301 * @see org.modeshape.graph.observe.Observable#register(org.modeshape.graph.observe.Observer)
302 */
303 @Override
304 public boolean register( Observer observer ) {
305 return observers.register(observer);
306 }
307
308 /**
309 * {@inheritDoc}
310 *
311 * @see org.modeshape.graph.observe.Observable#unregister(org.modeshape.graph.observe.Observer)
312 */
313 @Override
314 public boolean unregister( Observer observer ) {
315 return observers.unregister(observer);
316 }
317
318 /**
319 * {@inheritDoc}
320 *
321 * @see org.modeshape.graph.observe.ObservationBus#hasObservers()
322 */
323 @Override
324 public boolean hasObservers() {
325 return !observers.isEmpty();
326 }
327
328 /**
329 * Return whether this bus has been {@link #start() started} and not yet {@link #shutdown() shut down}.
330 *
331 * @return true if {@link #start()} has been called but {@link #shutdown()} has not, or false otherwise
332 */
333 public boolean isStarted() {
334 return channel != null;
335 }
336
337 /**
338 * {@inheritDoc}
339 *
340 * @see org.modeshape.graph.observe.ObservationBus#shutdown()
341 */
342 @Override
343 public synchronized void shutdown() {
344 if (channel != null) {
345 // Mark this as not accepting any more ...
346 isOpen.set(false);
347 try {
348 // Disconnect from the channel and close it ...
349 channel.removeChannelListener(listener);
350 channel.setReceiver(null);
351 channel.close();
352 } finally {
353 channel = null;
354 // Now that we're not receiving any more messages, shut down the list of observers ...
355 observers.shutdown();
356 }
357 }
358 }
359
360 protected static byte[] serialize( Changes changes ) throws Exception {
361 return Util.objectToByteBuffer(changes);
362 }
363
364 protected static Changes deserialize( byte[] data ) throws Exception {
365 return (Changes)Util.objectFromByteBuffer(data);
366 }
367
368 protected class Receiver implements org.jgroups.Receiver {
369 private byte[] state;
370
371 /**
372 * {@inheritDoc}
373 *
374 * @see org.jgroups.MembershipListener#block()
375 */
376 @Override
377 public void block() {
378 isOpen.set(false);
379 }
380
381 /**
382 * {@inheritDoc}
383 *
384 * @see org.jgroups.MessageListener#receive(org.jgroups.Message)
385 */
386 @Override
387 public void receive( Message message ) {
388 if (!observers.isEmpty()) {
389 // We have at least one observer ...
390 try {
391 // Deserialize the changes ...
392 Changes changes = deserialize(message.getBuffer());
393 // and broadcast to all of our observers ...
394 observers.broadcast(changes);
395 if (LOGGER.isTraceEnabled()) {
396 LOGGER.trace("Received on cluster '{0}' {1} changes in source '{2}' made by {3} from process '{4}' at {5}",
397 getClusterName(),
398 changes.getChangeRequests().size(),
399 changes.getSourceName(),
400 changes.getUserName(),
401 changes.getProcessId(),
402 changes.getTimestamp());
403 }
404 } catch (Exception e) {
405 // Something went wrong here (this should not happen) ...
406 String msg = ClusteringI18n.errorDeserializingChanges.text(getClusterName());
407 throw new SystemFailureException(msg, e);
408 }
409 }
410 }
411
412 /**
413 * {@inheritDoc}
414 *
415 * @see org.jgroups.MessageListener#getState()
416 */
417 @Override
418 public byte[] getState() {
419 return state;
420 }
421
422 /**
423 * {@inheritDoc}
424 *
425 * @see org.jgroups.MessageListener#setState(byte[])
426 */
427 @Override
428 public void setState( byte[] state ) {
429 this.state = state;
430 }
431
432 /**
433 * {@inheritDoc}
434 *
435 * @see org.jgroups.MembershipListener#suspect(org.jgroups.Address)
436 */
437 @Override
438 public void suspect( Address suspectedMbr ) {
439 LOGGER.error(ClusteringI18n.memberOfClusterIsSuspect, getClusterName(), suspectedMbr);
440 }
441
442 /**
443 * {@inheritDoc}
444 *
445 * @see org.jgroups.MembershipListener#viewAccepted(org.jgroups.View)
446 */
447 @Override
448 public void viewAccepted( View newView ) {
449 LOGGER.trace("Members of '{0}' cluster have changed: {1}", getClusterName(), newView);
450 if (newView.getMembers().size() > 1) {
451 if (multipleAddressesInCluster.compareAndSet(false, true)) {
452 LOGGER.debug("There are now multiple members of cluster '{0}'; changes will be propagated throughout the cluster",
453 getClusterName());
454 }
455 } else {
456 if (multipleAddressesInCluster.compareAndSet(true, false)) {
457 LOGGER.debug("There is only one member of cluster '{0}'; changes will be propagated locally only",
458 getClusterName());
459 }
460 }
461 }
462 }
463
464 protected class Listener implements ChannelListener {
465 /**
466 * {@inheritDoc}
467 *
468 * @see org.jgroups.ChannelListener#channelClosed(org.jgroups.Channel)
469 */
470 @Override
471 public void channelClosed( Channel channel ) {
472 isOpen.set(false);
473 }
474
475 /**
476 * {@inheritDoc}
477 *
478 * @see org.jgroups.ChannelListener#channelConnected(org.jgroups.Channel)
479 */
480 @Override
481 public void channelConnected( Channel channel ) {
482 isOpen.set(true);
483 }
484
485 /**
486 * {@inheritDoc}
487 *
488 * @see org.jgroups.ChannelListener#channelDisconnected(org.jgroups.Channel)
489 */
490 @Override
491 public void channelDisconnected( Channel channel ) {
492 isOpen.set(false);
493 }
494
495 /**
496 * {@inheritDoc}
497 *
498 * @see org.jgroups.ChannelListener#channelReconnected(org.jgroups.Address)
499 */
500 @Override
501 public void channelReconnected( Address addr ) {
502 isOpen.set(true);
503 }
504
505 /**
506 * {@inheritDoc}
507 *
508 * @see org.jgroups.ChannelListener#channelShunned()
509 */
510 @Override
511 public void channelShunned() {
512 isOpen.set(false);
513 }
514 }
515 }