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.gae;
022
023 import java.io.Serializable;
024 import java.util.ArrayList;
025 import java.util.Collection;
026 import java.util.HashMap;
027 import java.util.List;
028 import java.util.Map;
029
030 import org.granite.gravity.AsyncHttpContext;
031 import org.granite.gravity.AsyncPublishedMessage;
032 import org.granite.gravity.Channel;
033 import org.granite.gravity.MessagePublishingException;
034 import org.granite.gravity.MessageReceivingException;
035 import org.granite.gravity.Subscription;
036 import org.granite.logging.Logger;
037
038 import com.google.appengine.api.memcache.Expiration;
039 import com.google.appengine.api.memcache.MemcacheService;
040 import com.google.appengine.api.memcache.MemcacheServiceFactory;
041
042 import flex.messaging.messages.AsyncMessage;
043 import flex.messaging.messages.Message;
044
045 /**
046 * @author William DRAI
047 */
048 public class GAEChannel implements Channel, Serializable {
049
050 private static final long serialVersionUID = 5129029435795219401L;
051
052 private static final Logger log = Logger.getLogger(GAEChannel.class);
053
054 static final String MSG_COUNT_PREFIX = "org.granite.gravity.channel.msgCount.";
055 static final String MSG_PREFIX = "org.granite.gravity.channel.msg.";
056
057 private static MemcacheService gaeCache = MemcacheServiceFactory.getMemcacheService();
058
059 protected final String id;
060 protected final String clientType;
061 protected final GAEGravity gravity;
062 protected final GAEChannelFactory factory;
063
064 private final Map<String, Subscription> subscriptions = new HashMap<String, Subscription>();
065 private final long expiration;
066
067
068 GAEChannel(GAEGravity gravity, String id, GAEChannelFactory factory, String clientType) {
069 if (id == null)
070 throw new NullPointerException("id cannot be null");
071
072 this.id = id;
073 this.clientType = clientType;
074 this.factory = factory;
075 this.gravity = gravity;
076 this.expiration = gravity.getGravityConfig().getChannelIdleTimeoutMillis();
077 }
078
079 public String getId() {
080 return id;
081 }
082
083 public String getClientType() {
084 return clientType;
085 }
086
087 public GAEChannelFactory getFactory() {
088 return factory;
089 }
090
091 public GAEGravity getGravity() {
092 return gravity;
093 }
094
095 private Long msgCount() {
096 return (Long)gaeCache.get(MSG_COUNT_PREFIX + id);
097 }
098
099
100 public void close() {
101 }
102
103 public void destroy() {
104 Long msgCount = msgCount();
105 if (msgCount != null) {
106 List<Object> list = new ArrayList<Object>();
107 list.add(MSG_COUNT_PREFIX + id);
108 for (long i = 0; i < msgCount; i++)
109 list.add(MSG_PREFIX + id + "#" + i);
110 gaeCache.deleteAll(list);
111 }
112 this.subscriptions.clear();
113 }
114
115
116 public void publish(AsyncPublishedMessage message) throws MessagePublishingException {
117 message.publish(this);
118 }
119
120 public void receive(AsyncMessage message) throws MessageReceivingException {
121 log.debug("Publish message to channel %s", id);
122 // System.err.println("Publish messages to channel " + id);
123 synchronized (this) {
124 Long msgCount = msgCount();
125 gaeCache.put(MSG_PREFIX + id + "#" + msgCount, message, Expiration.byDeltaMillis((int)expiration));
126 gaeCache.increment(MSG_COUNT_PREFIX + id, 1);
127 }
128 }
129
130 public List<Message> takeMessages() {
131 log.debug("Try to take messages for channel %s", id);
132 // System.err.println("Try to take messages for channel " + id);
133 synchronized (this) {
134 Long msgCount = msgCount();
135 if (msgCount == null || msgCount == 0)
136 return null;
137
138 log.debug("Taking %s messages", msgCount);
139 // System.err.println("Taking " + msgCount + " messages");
140 List<Object> list = new ArrayList<Object>();
141 for (int i = 0; i < msgCount; i++)
142 list.add(MSG_PREFIX + id + "#" + i);
143 Map<Object, Object> msgs = gaeCache.getAll(list);
144 List<Message> messages = new ArrayList<Message>();
145 for (int i = 0; i < msgCount; i++) {
146 Message msg = (Message)msgs.get(list.get(i));
147 if (msg != null)
148 messages.add(msg);
149 }
150
151 gaeCache.deleteAll(list);
152 gaeCache.put(MSG_COUNT_PREFIX + id, 0L, Expiration.byDeltaMillis((int)expiration));
153
154 return messages.isEmpty() ? null : messages;
155 }
156 }
157
158
159 public Subscription addSubscription(String destination, String subTopicId, String subscriptionId, boolean noLocal) {
160 Subscription subscription = new Subscription(this, destination, subTopicId, subscriptionId, noLocal);
161 subscriptions.put(subscriptionId, subscription);
162 return subscription;
163 }
164
165 public Collection<Subscription> getSubscriptions() {
166 return subscriptions.values();
167 }
168
169 public Subscription removeSubscription(String subscriptionId) {
170 return subscriptions.remove(subscriptionId);
171 }
172
173
174 @Override
175 public boolean equals(Object obj) {
176 return (obj instanceof GAEChannel && id.equals(((GAEChannel)obj).id));
177 }
178
179 @Override
180 public int hashCode() {
181 return id.hashCode();
182 }
183
184 @Override
185 public String toString() {
186 return getClass().getName() + " {id=" + id + ", subscriptions=" + subscriptions + "}";
187 }
188
189
190 public boolean hasPublishedMessage() {
191 return false;
192 }
193
194 public boolean runPublish() {
195 return false;
196 }
197
198 public boolean hasReceivedMessage() {
199 return false;
200 }
201
202 public boolean runReceive() {
203 return false;
204 }
205
206 public boolean runReceived(AsyncHttpContext asyncHttpContext) {
207 return false;
208 }
209
210 public void run() {
211 }
212 }