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.tide.data;
022
023 import java.util.Map;
024 import java.util.Map.Entry;
025
026 import org.granite.clustering.DistributedData;
027 import org.granite.clustering.DistributedDataFactory;
028 import org.granite.context.GraniteContext;
029 import org.granite.gravity.Channel;
030 import org.granite.gravity.Gravity;
031 import org.granite.gravity.GravityManager;
032 import org.granite.logging.Logger;
033 import org.granite.messaging.webapp.ServletGraniteContext;
034
035 import flex.messaging.messages.AsyncMessage;
036 import flex.messaging.messages.CommandMessage;
037 import flex.messaging.messages.ErrorMessage;
038 import flex.messaging.messages.Message;
039
040
041 /**
042 * Default implementation for data update dispatchers using the Gravity API to dispatch updates.
043 *
044 * @see DataDispatcher
045 * @see DataContext
046 *
047 * @author William Drai
048 */
049 public class DefaultDataDispatcher extends AbstractDataDispatcher {
050
051 private static final Logger log = Logger.getLogger(DefaultDataDispatcher.class);
052
053
054 private Gravity gravity = null;
055
056
057 public DefaultDataDispatcher(Gravity gravity, String topicName, Class<? extends DataTopicParams> dataTopicParamsClass) {
058 super(topicName, dataTopicParamsClass);
059
060 GraniteContext graniteContext = GraniteContext.getCurrentInstance();
061 if (gravity == null && (graniteContext == null || !(graniteContext instanceof ServletGraniteContext)))
062 return;
063
064 DistributedDataFactory distributedDataFactory = graniteContext.getGraniteConfig().getDistributedDataFactory();
065 DistributedData gdd = distributedDataFactory.getInstance();
066 if (gdd != null) {
067 this.gravity = GravityManager.getGravity(((ServletGraniteContext)graniteContext).getServletContext());
068
069 if (this.gravity == null) {
070 log.debug("Gravity not found or HTTP session not found, data dispatch disabled");
071 return;
072 }
073
074 clientId = gdd.getDestinationClientId(topicName);
075 subscriptionId = gdd.getDestinationSubscriptionId(topicName);
076 sessionId = graniteContext.getSessionId();
077 }
078 else {
079 if (gravity == null) {
080 log.debug("Gravity not defined, data dispatch disabled");
081 return;
082 }
083
084 this.gravity = gravity;
085 this.sessionId = DataDispatcher.SERVER_DISPATCHER_GDS_SESSION_ID;
086 }
087
088 enabled = true;
089 }
090
091
092 @Override
093 protected void changeDataSelector(String dataSelector) {
094 DistributedDataFactory distributedDataFactory = GraniteContext.getCurrentInstance().getGraniteConfig().getDistributedDataFactory();
095 DistributedData gdd = distributedDataFactory.getInstance();
096 if (gdd != null) {
097 String clientId = gdd.getDestinationClientId(topicName);
098 String subscriptionId = gdd.getDestinationSubscriptionId(topicName);
099
100 if (clientId != null) {
101 CommandMessage message = new CommandMessage();
102 message.setClientId(clientId);
103 message.setHeader(AsyncMessage.DESTINATION_CLIENT_ID_HEADER, subscriptionId);
104 message.setHeader(AsyncMessage.SUBTOPIC_HEADER, TIDE_DATA_SUBTOPIC);
105 message.setDestination(topicName);
106 message.setOperation(CommandMessage.SUBSCRIBE_OPERATION);
107
108 message.setHeader(CommandMessage.SELECTOR_HEADER, dataSelector);
109
110 gravity.handleMessage(null, message, true);
111
112 log.debug("Topic %s data selector changed: %s", topicName, dataSelector);
113 }
114 }
115 }
116
117 @Override
118 public void publishUpdate(Map<String, String> params, Object body) {
119 AsyncMessage message = new AsyncMessage();
120 message.setDestination(topicName);
121 for (Entry<String, String> hh : params.entrySet())
122 message.setHeader(hh.getKey(), hh.getValue());
123 message.setBody(body);
124
125 Message resultMessage = null;
126 if (clientId != null) {
127 Channel channel = gravity.getChannel(null, clientId);
128 message.setClientId(clientId);
129 resultMessage = gravity.publishMessage(channel, message);
130 }
131 else
132 resultMessage = gravity.publishMessage(message);
133
134 if (resultMessage instanceof ErrorMessage)
135 log.error("Could not dispatch data update on topic %s, message %s", topicName, resultMessage.toString());
136 else
137 log.debug("Data message dispatched on topic %s", topicName);
138 }
139 }