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.ArrayList;
024 import java.util.HashMap;
025 import java.util.List;
026 import java.util.Map;
027 import java.util.Map.Entry;
028
029 import org.granite.clustering.DistributedData;
030 import org.granite.context.GraniteContext;
031 import org.granite.logging.Logger;
032
033 import flex.messaging.messages.AsyncMessage;
034
035
036 /**
037 * Base implementation for data update dispatchers.
038 * It should be built at beginning of each request during initialization of <code>DataContext</code>.
039 * The dispatch is a three step process :
040 *
041 * <ul>
042 * <li>Initialization in the constructor</li>
043 * <li><code>observe()</code> builds the server selector depending on the data that are processed</li>
044 * <li><code>publish()</code> handles the actual publishing</li>
045 * </ul>
046 *
047 * Actual implementations should only override <code>changeDataSelector</code> and <code>publishUpdate</code>/
048 *
049 * @see DataDispatcher
050 * @see DataContext
051 *
052 * @author William Drai
053 */
054 public abstract class AbstractDataDispatcher implements DataDispatcher {
055
056 private static final Logger log = Logger.getLogger(AbstractDataDispatcher.class);
057
058
059 protected boolean enabled;
060 protected String topicName = null;
061 protected DataTopicParams paramsProvider = null;
062 protected String sessionId = null;
063 protected String clientId = null;
064 protected String subscriptionId = null;
065
066
067 public AbstractDataDispatcher(String topicName, Class<? extends DataTopicParams> dataTopicParamsClass) {
068 this.topicName = topicName;
069
070 try {
071 paramsProvider = dataTopicParamsClass.newInstance();
072 }
073 catch (Exception e) {
074 log.error("Could not instantiate class " + dataTopicParamsClass, e);
075 }
076 }
077
078
079 public void observe() {
080 // Prepare the selector even if we are not yet subscribed
081 DataObserveParams params = null;
082 if (paramsProvider != null) {
083 // Collect selector parameters from component
084 params = new DataObserveParams();
085 paramsProvider.observes(params);
086 }
087
088 // Ensure that the current Gravity consumer listens about this data topic and params
089 GraniteContext graniteContext = GraniteContext.getCurrentInstance();
090 if (graniteContext == null)
091 return;
092
093 DistributedData gdd = graniteContext.getGraniteConfig().getDistributedDataFactory().getInstance();
094 if (gdd == null)
095 return; // Session expired
096
097 List<DataObserveParams> selectors = DataObserveParams.fromSerializableForm(gdd.getDestinationDataSelectors(topicName));
098 List<DataObserveParams> newSelectors = new ArrayList<DataObserveParams>(selectors);
099
100 boolean dataSelectorChanged = false;
101 String dataSelector = gdd.getDestinationSelector(topicName);
102 if (params != null) {
103 String newDataSelector = params.updateDataSelector(dataSelector, newSelectors);
104 dataSelectorChanged = !newDataSelector.equals(dataSelector);
105 if (dataSelectorChanged) {
106 log.debug("Data selector changed: %s", newDataSelector);
107 gdd.setDestinationSelector(topicName, newDataSelector);
108 dataSelector = newDataSelector;
109 }
110 }
111
112 if (!DataObserveParams.containsSame(selectors, newSelectors)) {
113 log.debug("Selectors changed: %s", newSelectors);
114 gdd.setDestinationDataSelectors(topicName, DataObserveParams.toSerializableForm(newSelectors));
115 }
116
117 if (!enabled)
118 return;
119
120 if (dataSelectorChanged)
121 changeDataSelector(dataSelector);
122 }
123
124 protected abstract void changeDataSelector(String dataSelector);
125
126
127 public void publish(Object[][] dataUpdates) {
128 if (!enabled)
129 return;
130
131 try {
132 Map<Map<String, String>, List<Object>> updates = new HashMap<Map<String, String>, List<Object>>();
133 if (paramsProvider != null) {
134 for (Object[] dataUpdate : dataUpdates) {
135 DataPublishParams params = new DataPublishParams();
136 paramsProvider.publishes(params, dataUpdate[1]);
137
138 Map<String, String> headers = params.getHeaders();
139 List<Object> list = updates.get(headers);
140 if (list == null) {
141 list = new ArrayList<Object>();
142 updates.put(headers, list);
143 }
144 list.add(dataUpdate);
145 }
146 }
147
148 for (Entry<Map<String, String>, List<Object>> me : updates.entrySet()) {
149 Map<String, String> headers = new HashMap<String, String>(me.getKey());
150 headers.put(AsyncMessage.SUBTOPIC_HEADER, TIDE_DATA_SUBTOPIC);
151 headers.put(GDS_SESSION_ID, sessionId);
152 headers.put(TIDE_DATA_TYPE_KEY, TIDE_DATA_TYPE_VALUE);
153 publishUpdate(headers, me.getValue().toArray());
154 }
155 }
156 catch (Exception e) {
157 log.error(e, "Could not publish data update on topic %s", topicName);
158 }
159 }
160
161 protected abstract void publishUpdate(Map<String, String> params, Object body);
162 }