001 /*
002 GRANITE DATA SERVICES
003 Copyright (C) 2011 GRANITE DATA SERVICES S.A.S.
004
005 This file is part of Granite Data Services.
006
007 Granite Data Services is free software; you can redistribute it and/or modify
008 it under the terms of the GNU Library General Public License as published by
009 the Free Software Foundation; either version 2 of the License, or (at your
010 option) any later version.
011
012 Granite Data Services is distributed in the hope that it will be useful, but
013 WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
014 FITNESS FOR A PARTICULAR PURPOSE. See the GNU Library General Public License
015 for more details.
016
017 You should have received a copy of the GNU Library General Public License
018 along with this library; if not, see <http://www.gnu.org/licenses/>.
019 */
020
021 package org.granite.gravity;
022
023 import java.io.Serializable;
024 import java.util.Date;
025 import java.util.HashMap;
026 import java.util.Map;
027 import java.util.Timer;
028 import java.util.TimerTask;
029 import java.util.concurrent.ConcurrentHashMap;
030
031 import javax.management.ObjectName;
032
033 import org.granite.clustering.DistributedData;
034 import org.granite.config.GraniteConfig;
035 import org.granite.config.flex.Destination;
036 import org.granite.config.flex.ServicesConfig;
037 import org.granite.context.GraniteContext;
038 import org.granite.context.SimpleGraniteContext;
039 import org.granite.gravity.adapters.AdapterFactory;
040 import org.granite.gravity.adapters.ServiceAdapter;
041 import org.granite.gravity.security.GravityDestinationSecurizer;
042 import org.granite.gravity.security.GravityInvocationContext;
043 import org.granite.jmx.MBeanServerLocator;
044 import org.granite.jmx.OpenMBean;
045 import org.granite.logging.Logger;
046 import org.granite.messaging.amf.process.AMF3MessageInterceptor;
047 import org.granite.messaging.service.security.SecurityService;
048 import org.granite.messaging.service.security.SecurityServiceException;
049 import org.granite.messaging.webapp.ServletGraniteContext;
050 import org.granite.util.TypeUtil;
051 import org.granite.util.UUIDUtil;
052
053 import flex.messaging.messages.AcknowledgeMessage;
054 import flex.messaging.messages.AsyncMessage;
055 import flex.messaging.messages.CommandMessage;
056 import flex.messaging.messages.ErrorMessage;
057 import flex.messaging.messages.Message;
058
059 /**
060 * @author William DRAI
061 * @author Franck WOLFF
062 */
063 public class DefaultGravity implements Gravity, DefaultGravityMBean {
064
065 ///////////////////////////////////////////////////////////////////////////
066 // Fields.
067
068 private static final Logger log = Logger.getLogger(Gravity.class);
069
070 private final Map<String, Object> applicationMap = new HashMap<String, Object>();
071 private final ConcurrentHashMap<String, TimeChannel<?>> channels = new ConcurrentHashMap<String, TimeChannel<?>>();
072
073 private GravityConfig gravityConfig = null;
074 private ServicesConfig servicesConfig = null;
075 private GraniteConfig graniteConfig = null;
076
077 private Channel serverChannel = null;
078 private AdapterFactory adapterFactory = null;
079 private GravityPool gravityPool = null;
080
081 private Timer channelsTimer;
082 private boolean started;
083
084 ///////////////////////////////////////////////////////////////////////////
085 // Constructor.
086
087 public DefaultGravity(GravityConfig gravityConfig, ServicesConfig servicesConfig, GraniteConfig graniteConfig) {
088 if (gravityConfig == null || servicesConfig == null || graniteConfig == null)
089 throw new NullPointerException("All arguments must be non null.");
090
091 this.gravityConfig = gravityConfig;
092 this.servicesConfig = servicesConfig;
093 this.graniteConfig = graniteConfig;
094 }
095
096 ///////////////////////////////////////////////////////////////////////////
097 // Properties.
098
099 public GravityConfig getGravityConfig() {
100 return gravityConfig;
101 }
102
103 public ServicesConfig getServicesConfig() {
104 return servicesConfig;
105 }
106
107 public GraniteConfig getGraniteConfig() {
108 return graniteConfig;
109 }
110
111 public boolean isStarted() {
112 return started;
113 }
114
115 public ServiceAdapter getServiceAdapter(String messageType, String destinationId) {
116 return adapterFactory.getServiceAdapter(messageType, destinationId);
117 }
118
119 ///////////////////////////////////////////////////////////////////////////
120 // Starting/stopping.
121
122 public void start() throws Exception {
123 log.info("Starting Gravity...");
124 synchronized (this) {
125 if (!started) {
126 adapterFactory = new AdapterFactory(this);
127 internalStart();
128 serverChannel = new ServerChannel(this, ServerChannel.class.getName(), null, null);
129 started = true;
130 }
131 }
132 log.info("Gravity successfully started.");
133 }
134
135 protected void internalStart() {
136 gravityPool = new GravityPool(gravityConfig);
137 channelsTimer = new Timer();
138
139 if (graniteConfig.isRegisterMBeans()) {
140 try {
141 ObjectName name = new ObjectName("org.graniteds:type=Gravity,context=" + graniteConfig.getMBeanContextName());
142 log.info("Registering MBean: %s", name);
143 OpenMBean mBean = OpenMBean.createMBean(this);
144 MBeanServerLocator.getInstance().register(mBean, name, true);
145 }
146 catch (Exception e) {
147 log.error(e, "Could not register Gravity MBean for context: %s", graniteConfig.getMBeanContextName());
148 }
149 }
150 }
151
152 public void restart() throws Exception {
153 synchronized (this) {
154 stop();
155 start();
156 }
157 }
158
159 public void reconfigure(GravityConfig gravityConfig, GraniteConfig graniteConfig) {
160 this.gravityConfig = gravityConfig;
161 this.graniteConfig = graniteConfig;
162 if (gravityPool != null)
163 gravityPool.reconfigure(gravityConfig);
164 }
165
166 public void stop() throws Exception {
167 stop(true);
168 }
169
170 public void stop(boolean now) throws Exception {
171 log.info("Stopping Gravity (now=%s)...", now);
172 synchronized (this) {
173 if (adapterFactory != null) {
174 try {
175 adapterFactory.stopAll();
176 } catch (Exception e) {
177 log.error(e, "Error while stopping adapter factory");
178 }
179 adapterFactory = null;
180 }
181
182 if (serverChannel != null) {
183 try {
184 removeChannel(serverChannel.getId());
185 } catch (Exception e) {
186 log.error(e, "Error while removing server channel: %s", serverChannel);
187 }
188 serverChannel = null;
189 }
190
191 if (channelsTimer != null) {
192 try {
193 channelsTimer.cancel();
194 } catch (Exception e) {
195 log.error(e, "Error while cancelling channels timer");
196 }
197 channelsTimer = null;
198 }
199
200 if (gravityPool != null) {
201 try {
202 if (now)
203 gravityPool.shutdownNow();
204 else
205 gravityPool.shutdown();
206 }
207 catch (Exception e) {
208 log.error(e, "Error while stopping thread pool");
209 }
210 gravityPool = null;
211 }
212
213 started = false;
214 }
215 log.info("Gravity sucessfully stopped.");
216 }
217
218 ///////////////////////////////////////////////////////////////////////////
219 // GravityMBean attributes implementation.
220
221 public String getGravityFactoryName() {
222 return gravityConfig.getGravityFactory();
223 }
224
225 public long getChannelIdleTimeoutMillis() {
226 return gravityConfig.getChannelIdleTimeoutMillis();
227 }
228 public void setChannelIdleTimeoutMillis(long channelIdleTimeoutMillis) {
229 gravityConfig.setChannelIdleTimeoutMillis(channelIdleTimeoutMillis);
230 }
231
232 public boolean isRetryOnError() {
233 return gravityConfig.isRetryOnError();
234 }
235 public void setRetryOnError(boolean retryOnError) {
236 gravityConfig.setRetryOnError(retryOnError);
237 }
238
239 public long getLongPollingTimeoutMillis() {
240 return gravityConfig.getLongPollingTimeoutMillis();
241 }
242 public void setLongPollingTimeoutMillis(long longPollingTimeoutMillis) {
243 gravityConfig.setLongPollingTimeoutMillis(longPollingTimeoutMillis);
244 }
245
246 public int getMaxMessagesQueuedPerChannel() {
247 return gravityConfig.getMaxMessagesQueuedPerChannel();
248 }
249 public void setMaxMessagesQueuedPerChannel(int maxMessagesQueuedPerChannel) {
250 gravityConfig.setMaxMessagesQueuedPerChannel(maxMessagesQueuedPerChannel);
251 }
252
253 public long getReconnectIntervalMillis() {
254 return gravityConfig.getReconnectIntervalMillis();
255 }
256
257 public int getReconnectMaxAttempts() {
258 return gravityConfig.getReconnectMaxAttempts();
259 }
260
261 public int getCorePoolSize() {
262 if (gravityPool != null)
263 return gravityPool.getCorePoolSize();
264 return gravityConfig.getCorePoolSize();
265 }
266
267 public void setCorePoolSize(int corePoolSize) {
268 gravityConfig.setCorePoolSize(corePoolSize);
269 if (gravityPool != null)
270 gravityPool.setCorePoolSize(corePoolSize);
271 }
272
273 public long getKeepAliveTimeMillis() {
274 if (gravityPool != null)
275 return gravityPool.getKeepAliveTimeMillis();
276 return gravityConfig.getKeepAliveTimeMillis();
277 }
278 public void setKeepAliveTimeMillis(long keepAliveTimeMillis) {
279 gravityConfig.setKeepAliveTimeMillis(keepAliveTimeMillis);
280 if (gravityPool != null)
281 gravityPool.setKeepAliveTimeMillis(keepAliveTimeMillis);
282 }
283
284 public int getMaximumPoolSize() {
285 if (gravityPool != null)
286 return gravityPool.getMaximumPoolSize();
287 return gravityConfig.getMaximumPoolSize();
288 }
289 public void setMaximumPoolSize(int maximumPoolSize) {
290 gravityConfig.setMaximumPoolSize(maximumPoolSize);
291 if (gravityPool != null)
292 gravityPool.setMaximumPoolSize(maximumPoolSize);
293 }
294
295 public int getQueueCapacity() {
296 if (gravityPool != null)
297 return gravityPool.getQueueCapacity();
298 return gravityConfig.getQueueCapacity();
299 }
300
301 public int getQueueRemainingCapacity() {
302 if (gravityPool != null)
303 return gravityPool.getQueueRemainingCapacity();
304 return gravityConfig.getQueueCapacity();
305 }
306
307 public int getQueueSize() {
308 if (gravityPool != null)
309 return gravityPool.getQueueSize();
310 return 0;
311 }
312
313 ///////////////////////////////////////////////////////////////////////////
314 // Channel's operations.
315
316 protected <C extends Channel> C createChannel(ChannelFactory<C> channelFactory, String clientId) {
317 C channel = null;
318 if (clientId != null) {
319 channel = getChannel(channelFactory, clientId);
320 if (channel != null)
321 return channel;
322 }
323
324 String clientType = GraniteContext.getCurrentInstance().getClientType();
325 channel = channelFactory.newChannel(UUIDUtil.randomUUID(), clientType);
326 TimeChannel<C> timeChannel = new TimeChannel<C>(channel);
327 for (int i = 0; channels.putIfAbsent(channel.getId(), timeChannel) != null; i++) {
328 if (i >= 10)
329 throw new RuntimeException("Could not find random new clientId after 10 iterations");
330 channel.destroy();
331 channel = channelFactory.newChannel(UUIDUtil.randomUUID(), clientType);
332 timeChannel = new TimeChannel<C>(channel);
333 }
334
335 String channelId = channel.getId();
336
337 // Save channel id in distributed data (clustering).
338 try {
339 DistributedData gdd = graniteConfig.getDistributedDataFactory().getInstance();
340 if (gdd != null) {
341 log.debug("Saving channel id in distributed data: %s", channelId);
342 gdd.addChannelId(channelId, channelFactory.getClass().getName());
343 }
344 }
345 catch (Exception e) {
346 log.error(e, "Could not add channel id in distributed data: %s", channelId);
347 }
348
349 // Initialize timer task.
350 access(channelId);
351
352 return channel;
353 }
354
355 @SuppressWarnings("unchecked")
356 public <C extends Channel> C getChannel(ChannelFactory<C> channelFactory, String clientId) {
357 if (clientId == null)
358 return null;
359
360 TimeChannel<C> timeChannel = (TimeChannel<C>)channels.get(clientId);
361 if (timeChannel == null) {
362 // Look for existing channel id/subscriptions in distributed data (clustering).
363 try {
364 DistributedData gdd = graniteConfig.getDistributedDataFactory().getInstance();
365 if (gdd != null && gdd.hasChannelId(clientId)) {
366 log.debug("Found channel id in distributed data: %s", clientId);
367 String channelFactoryClassName = gdd.getChannelFactoryClassName(clientId);
368 channelFactory = TypeUtil.newInstance(channelFactoryClassName, ChannelFactory.class);
369 String clientType = GraniteContext.getCurrentInstance().getClientType();
370 C channel = channelFactory.newChannel(clientId, clientType);
371 timeChannel = new TimeChannel<C>(channel);
372 if (channels.putIfAbsent(clientId, timeChannel) == null) {
373 for (CommandMessage subscription : gdd.getSubscriptions(clientId)) {
374 log.debug("Resubscribing channel: %s - %s", clientId, subscription);
375 handleSubscribeMessage(channelFactory, subscription, false);
376 }
377 access(clientId);
378 }
379 }
380 }
381 catch (Exception e) {
382 log.error(e, "Could not recreate channel/subscriptions from distributed data: %s", clientId);
383 }
384 }
385
386 return (timeChannel != null ? timeChannel.getChannel() : null);
387 }
388
389 public Channel removeChannel(String channelId) {
390 if (channelId == null)
391 return null;
392
393 // Remove existing channel id/subscriptions in distributed data (clustering).
394 try {
395 DistributedData gdd = graniteConfig.getDistributedDataFactory().getInstance();
396 if (gdd != null) {
397 log.debug("Removing channel id from distributed data: %s", channelId);
398 gdd.removeChannelId(channelId);
399 }
400 }
401 catch (Exception e) {
402 log.error(e, "Could not remove channel id from distributed data: %s", channelId);
403 }
404
405 TimeChannel<?> timeChannel = channels.remove(channelId);
406 Channel channel = null;
407 if (timeChannel != null) {
408 try {
409 if (timeChannel.getTimerTask() != null)
410 timeChannel.getTimerTask().cancel();
411 }
412 catch (Exception e) {
413 // Should never happen...
414 }
415
416 channel = timeChannel.getChannel();
417
418 try {
419 for (Subscription subscription : channel.getSubscriptions()) {
420 try {
421 Message message = subscription.getUnsubscribeMessage();
422 handleMessage(channel.getFactory(), message, true);
423 }
424 catch (Exception e) {
425 log.error(e, "Error while unsubscribing channel: %s from subscription: %s", channel, subscription);
426 }
427 }
428 }
429 finally {
430 channel.destroy();
431 }
432 }
433
434 return channel;
435 }
436
437 public boolean access(String channelId) {
438 if (channelId != null) {
439 TimeChannel<?> timeChannel = channels.get(channelId);
440 if (timeChannel != null) {
441 synchronized (timeChannel) {
442 TimerTask timerTask = timeChannel.getTimerTask();
443 if (timerTask != null) {
444 log.debug("Canceling TimerTask: %s", timerTask);
445 timerTask.cancel();
446 timeChannel.setTimerTask(null);
447 }
448
449 timerTask = new ChannelTimerTask(this, channelId);
450 timeChannel.setTimerTask(timerTask);
451
452 long timeout = gravityConfig.getChannelIdleTimeoutMillis();
453 log.debug("Scheduling TimerTask: %s for %s ms.", timerTask, timeout);
454 channelsTimer.schedule(timerTask, timeout);
455 return true;
456 }
457 }
458 }
459 return false;
460 }
461
462 public void execute(AsyncChannelRunner runner) {
463 if (gravityPool == null) {
464 runner.reset();
465 throw new NullPointerException("Gravity not started or pool disabled");
466 }
467 gravityPool.execute(runner);
468 }
469
470 public boolean cancel(AsyncChannelRunner runner) {
471 if (gravityPool == null) {
472 runner.reset();
473 throw new NullPointerException("Gravity not started or pool disabled");
474 }
475 return gravityPool.remove(runner);
476 }
477
478 ///////////////////////////////////////////////////////////////////////////
479 // Incoming message handling.
480
481 public Message handleMessage(final ChannelFactory<?> channelFactory, Message message) {
482 return handleMessage(channelFactory, message, false);
483 }
484
485 public Message handleMessage(final ChannelFactory<?> channelFactory, final Message message, boolean skipInterceptor) {
486
487 AMF3MessageInterceptor interceptor = null;
488 if (!skipInterceptor)
489 interceptor = GraniteContext.getCurrentInstance().getGraniteConfig().getAmf3MessageInterceptor();
490
491 Message reply = null;
492 boolean publish = false;
493
494 try {
495 if (interceptor != null)
496 interceptor.before(message);
497
498 if (message instanceof CommandMessage) {
499 CommandMessage command = (CommandMessage)message;
500
501 switch (command.getOperation()) {
502
503 case CommandMessage.LOGIN_OPERATION:
504 case CommandMessage.LOGOUT_OPERATION:
505 return handleSecurityMessage(command);
506
507 case CommandMessage.CLIENT_PING_OPERATION:
508 return handlePingMessage(channelFactory, command);
509 case CommandMessage.CONNECT_OPERATION:
510 return handleConnectMessage(channelFactory, command);
511 case CommandMessage.DISCONNECT_OPERATION:
512 return handleDisconnectMessage(channelFactory, command);
513 case CommandMessage.SUBSCRIBE_OPERATION:
514 return handleSubscribeMessage(channelFactory, command);
515 case CommandMessage.UNSUBSCRIBE_OPERATION:
516 return handleUnsubscribeMessage(channelFactory, command);
517
518 default:
519 throw new UnsupportedOperationException("Unsupported command operation: " + command);
520 }
521 }
522
523 reply = handlePublishMessage(channelFactory, (AsyncMessage)message);
524 publish = true;
525 }
526 finally {
527 if (interceptor != null)
528 interceptor.after(message, reply);
529 }
530
531 if (reply != null) {
532 GraniteContext context = GraniteContext.getCurrentInstance();
533 if (context.getSessionId() != null) {
534 reply.setHeader("org.granite.sessionId", context.getSessionId());
535 if (publish && context instanceof ServletGraniteContext && ((ServletGraniteContext)context).getSession(false) != null) {
536 long serverTime = new Date().getTime();
537 ((ServletGraniteContext)context).getSession().setAttribute(GraniteContext.SESSION_LAST_ACCESSED_TIME_KEY, serverTime);
538 reply.setHeader("org.granite.time", serverTime);
539 reply.setHeader("org.granite.sessionExp", ((ServletGraniteContext)context).getSession().getMaxInactiveInterval());
540 }
541 }
542 }
543
544 return reply;
545 }
546
547 ///////////////////////////////////////////////////////////////////////////
548 // Other Public API methods.
549
550 public GraniteContext initThread(String sessionId, String clientType) {
551 GraniteContext context = GraniteContext.getCurrentInstance();
552 if (context == null)
553 context = SimpleGraniteContext.createThreadInstance(graniteConfig, servicesConfig, sessionId, applicationMap, clientType);
554 return context;
555 }
556
557 public void releaseThread() {
558 GraniteContext.release();
559 }
560
561 public Message publishMessage(AsyncMessage message) {
562 return publishMessage(serverChannel, message);
563 }
564
565 public Message publishMessage(Channel fromChannel, AsyncMessage message) {
566 initThread(null, fromChannel != null ? fromChannel.getClientType() : serverChannel.getClientType());
567
568 return handlePublishMessage(null, message, fromChannel != null ? fromChannel : serverChannel);
569 }
570
571 private Message handlePingMessage(ChannelFactory<?> channelFactory, CommandMessage message) {
572
573 Channel channel = createChannel(channelFactory, (String)message.getClientId());
574
575 AsyncMessage reply = new AcknowledgeMessage(message);
576 reply.setClientId(channel.getId());
577 Map<String, Object> advice = new HashMap<String, Object>();
578 advice.put(RECONNECT_INTERVAL_MS_KEY, Long.valueOf(gravityConfig.getReconnectIntervalMillis()));
579 advice.put(RECONNECT_MAX_ATTEMPTS_KEY, Long.valueOf(gravityConfig.getReconnectMaxAttempts()));
580 reply.setBody(advice);
581 reply.setDestination(message.getDestination());
582
583 log.debug("handshake.handle: reply=%s", reply);
584
585 return reply;
586 }
587
588 private Message handleSecurityMessage(CommandMessage message) {
589 GraniteConfig config = GraniteContext.getCurrentInstance().getGraniteConfig();
590
591 Message response = null;
592
593 if (!config.hasSecurityService())
594 log.warn("Ignored security operation (no security settings in granite-config.xml): %s", message);
595 else if (!config.getSecurityService().acceptsContext())
596 log.info("Ignored security operation (security service does not handle this kind of granite context)", message);
597 else {
598 SecurityService securityService = config.getSecurityService();
599 try {
600 if (message.isLoginOperation())
601 securityService.login(message.getBody(), (String)message.getHeader(Message.CREDENTIALS_CHARSET_HEADER));
602 else
603 securityService.logout();
604 }
605 catch (Exception e) {
606 if (e instanceof SecurityServiceException)
607 log.debug(e, "Could not process security operation: %s", message);
608 else
609 log.error(e, "Could not process security operation: %s", message);
610 response = new ErrorMessage(message, e, true);
611 }
612 }
613
614 if (response == null) {
615 response = new AcknowledgeMessage(message, true);
616 // For SDK 2.0.1_Hotfix2.
617 if (message.isSecurityOperation())
618 response.setBody("success");
619 }
620
621 return response;
622 }
623
624 private Message handleConnectMessage(final ChannelFactory<?> channelFactory, CommandMessage message) {
625 Channel client = getChannel(channelFactory, (String)message.getClientId());
626
627 if (client == null)
628 return handleUnknownClientMessage(message);
629
630 return null;
631 }
632
633 private Message handleDisconnectMessage(final ChannelFactory<?> channelFactory, CommandMessage message) {
634 Channel client = getChannel(channelFactory, (String)message.getClientId());
635 if (client == null)
636 return handleUnknownClientMessage(message);
637
638 removeChannel(client.getId());
639
640 AcknowledgeMessage reply = new AcknowledgeMessage(message);
641 reply.setDestination(message.getDestination());
642 reply.setClientId(client.getId());
643 return reply;
644 }
645
646 private Message handleSubscribeMessage(final ChannelFactory<?> channelFactory, final CommandMessage message) {
647 return handleSubscribeMessage(channelFactory, message, true);
648 }
649
650 private Message handleSubscribeMessage(final ChannelFactory<?> channelFactory, final CommandMessage message, final boolean saveMessageInSession) {
651
652 final GraniteContext context = GraniteContext.getCurrentInstance();
653
654 // Get and check destination.
655 final Destination destination = context.getServicesConfig().findDestinationById(
656 message.getMessageRefType(),
657 message.getDestination()
658 );
659
660 if (destination == null)
661 return getInvalidDestinationError(message);
662
663
664 GravityInvocationContext invocationContext = new GravityInvocationContext(message, destination) {
665 @Override
666 public Object invoke() throws Exception {
667 // Subscribe...
668 Channel channel = getChannel(channelFactory, (String)message.getClientId());
669 if (channel == null)
670 return handleUnknownClientMessage(message);
671
672 String subscriptionId = (String)message.getHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER);
673 if (subscriptionId == null) {
674 subscriptionId = UUIDUtil.randomUUID();
675 message.setHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER, subscriptionId);
676 }
677
678 DistributedData gdd = graniteConfig.getDistributedDataFactory().getInstance();
679 if (gdd != null) {
680 if (Boolean.TRUE.toString().equals(destination.getProperties().get("session-selector"))) {
681 String selector = gdd.getDestinationSelector(destination.getId());
682 log.debug("Session selector found: %s", selector);
683 if (selector != null)
684 message.setHeader(CommandMessage.SELECTOR_HEADER, selector);
685 }
686 }
687
688 ServiceAdapter adapter = adapterFactory.getServiceAdapter(message);
689
690 AsyncMessage reply = (AsyncMessage)adapter.manage(channel, message);
691
692 postManage(channel);
693
694 if (saveMessageInSession && !(reply instanceof ErrorMessage)) {
695 // Save subscription message in distributed data (clustering).
696 try {
697 if (gdd != null) {
698 log.debug("Saving new subscription message for channel: %s - %s", channel.getId(), message);
699 gdd.addSubcription(channel.getId(), message);
700 }
701 }
702 catch (Exception e) {
703 log.error(e, "Could not add subscription in distributed data: %s - %s", channel.getId(), subscriptionId);
704 }
705 }
706
707 reply.setDestination(message.getDestination());
708 reply.setClientId(channel.getId());
709 reply.getHeaders().putAll(message.getHeaders());
710
711 if (gdd != null && message.getDestination() != null) {
712 gdd.setDestinationClientId(message.getDestination(), channel.getId());
713 gdd.setDestinationSubscriptionId(message.getDestination(), subscriptionId);
714 }
715
716 return reply;
717 }
718 };
719
720 // Check security 1 (destination).
721 if (destination.getSecurizer() instanceof GravityDestinationSecurizer) {
722 try {
723 ((GravityDestinationSecurizer)destination.getSecurizer()).canSubscribe(invocationContext);
724 }
725 catch (Exception e) {
726 return new ErrorMessage(message, e);
727 }
728 }
729
730 // Check security 2 (security service).
731 GraniteConfig config = context.getGraniteConfig();
732 try {
733 if (config.hasSecurityService() && config.getSecurityService().acceptsContext())
734 return (Message)config.getSecurityService().authorize(invocationContext);
735
736 return (Message)invocationContext.invoke();
737 }
738 catch (Exception e) {
739 return new ErrorMessage(message, e, true);
740 }
741 }
742
743 private Message handleUnsubscribeMessage(final ChannelFactory<?> channelFactory, CommandMessage message) {
744 Channel channel = getChannel(channelFactory, (String)message.getClientId());
745 if (channel == null)
746 return handleUnknownClientMessage(message);
747
748 AsyncMessage reply = null;
749
750 ServiceAdapter adapter = adapterFactory.getServiceAdapter(message);
751
752 reply = (AcknowledgeMessage)adapter.manage(channel, message);
753
754 postManage(channel);
755
756 if (!(reply instanceof ErrorMessage)) {
757 // Remove subscription message in distributed data (clustering).
758 try {
759 DistributedData gdd = graniteConfig.getDistributedDataFactory().getInstance();
760 if (gdd != null) {
761 String subscriptionId = (String)message.getHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER);
762 log.debug("Removing subscription message from channel info: %s - %s", channel.getId(), subscriptionId);
763 gdd.removeSubcription(channel.getId(), subscriptionId);
764 }
765 }
766 catch (Exception e) {
767 log.error(
768 e, "Could not remove subscription from distributed data: %s - %s",
769 channel.getId(), message.getHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER)
770 );
771 }
772 }
773
774 reply.setDestination(message.getDestination());
775 reply.setClientId(channel.getId());
776 reply.getHeaders().putAll(message.getHeaders());
777
778 return reply;
779 }
780
781 protected void postManage(Channel channel) {
782 }
783
784 private Message handlePublishMessage(final ChannelFactory<?> channelFactory, final AsyncMessage message) {
785 return handlePublishMessage(channelFactory, message, null);
786 }
787
788 private Message handlePublishMessage(final ChannelFactory<?> channelFactory, final AsyncMessage message, final Channel channel) {
789
790 GraniteContext context = GraniteContext.getCurrentInstance();
791
792 // Get and check destination.
793 Destination destination = context.getServicesConfig().findDestinationById(
794 message.getClass().getName(),
795 message.getDestination()
796 );
797
798 if (destination == null)
799 return getInvalidDestinationError(message);
800
801 if (message.getMessageId() == null)
802 message.setMessageId(UUIDUtil.randomUUID());
803 message.setTimestamp(System.currentTimeMillis());
804 if (channel != null)
805 message.setClientId(channel.getId());
806
807 GravityInvocationContext invocationContext = new GravityInvocationContext(message, destination) {
808 @Override
809 public Object invoke() throws Exception {
810 // Publish...
811 Channel fromChannel = channel;
812 if (fromChannel == null)
813 fromChannel = getChannel(channelFactory, (String)message.getClientId());
814 if (fromChannel == null)
815 return handleUnknownClientMessage(message);
816
817 ServiceAdapter adapter = adapterFactory.getServiceAdapter(message);
818
819 AsyncMessage reply = (AsyncMessage)adapter.invoke(fromChannel, message);
820
821 reply.setDestination(message.getDestination());
822 reply.setClientId(fromChannel.getId());
823
824 return reply;
825 }
826 };
827
828 // Check security 1 (destination).
829 if (destination.getSecurizer() instanceof GravityDestinationSecurizer) {
830 try {
831 ((GravityDestinationSecurizer)destination.getSecurizer()).canPublish(invocationContext);
832 }
833 catch (Exception e) {
834 return new ErrorMessage(message, e, true);
835 }
836 }
837
838 // Check security 2 (security service).
839 GraniteConfig config = context.getGraniteConfig();
840 try {
841 if (config.hasSecurityService() && config.getSecurityService().acceptsContext())
842 return (Message)config.getSecurityService().authorize(invocationContext);
843
844 return (Message)invocationContext.invoke();
845 }
846 catch (Exception e) {
847 return new ErrorMessage(message, e, true);
848 }
849 }
850
851 private Message handleUnknownClientMessage(Message message) {
852 ErrorMessage reply = new ErrorMessage(message, true);
853 reply.setFaultCode("Server.Call.UnknownClient");
854 reply.setFaultString("Unknown client");
855 return reply;
856 }
857
858 ///////////////////////////////////////////////////////////////////////////
859 // Utilities.
860
861 private ErrorMessage getInvalidDestinationError(Message message) {
862
863 String messageType = message.getClass().getName();
864 if (message instanceof CommandMessage)
865 messageType += '[' + ((CommandMessage)message).getMessageRefType() + ']';
866
867 ErrorMessage reply = new ErrorMessage(message, true);
868 reply.setFaultCode("Server.Messaging.InvalidDestination");
869 reply.setFaultString(
870 "No configured destination for id: " + message.getDestination() +
871 " and message type: " + messageType
872 );
873 return reply;
874 }
875
876 private static class ServerChannel extends AbstractChannel implements Serializable {
877
878 private static final long serialVersionUID = 1L;
879
880 public ServerChannel(Gravity gravity, String channelId, ChannelFactory<ServerChannel> factory, String clientType) {
881 super(gravity, channelId, factory, clientType);
882 }
883
884 @Override
885 public Gravity getGravity() {
886 return gravity;
887 }
888
889 public void close() {
890 }
891
892 @Override
893 public void receive(AsyncMessage message) throws MessageReceivingException {
894 }
895
896 @Override
897 protected boolean hasAsyncHttpContext() {
898 return false;
899 }
900
901 @Override
902 protected AsyncHttpContext acquireAsyncHttpContext() {
903 return null;
904 }
905
906 @Override
907 protected void releaseAsyncHttpContext(AsyncHttpContext context) {
908 }
909 }
910 }