ambidextrous/gemp-lotr
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

LongPollingSystem in place.

Browse Source
This commit is contained in:
marcins78
2013-01-09 12:51:18 +00:00
parent e2dfeba47a
commit 56ec2a04b9
5 changed files with 206 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LongPollingResource.java Normal file
View File

@@ -0,0 +1,7 @@
package com.gempukku.lotro.async;
public interface LongPollingResource {
public boolean isChanged();
public void process();
}

87
gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LongPollingSystem.java Normal file
View File

@@ -0,0 +1,87 @@
package com.gempukku.lotro.async;
import org.apache.log4j.Logger;
import java.util.concurrent.ConcurrentLinkedQueue;
public class LongPollingSystem {
private static Logger _log = Logger.getLogger(LongPollingSystem.class);
private ConcurrentLinkedQueue<LongPollingResource> _waitingActions = new ConcurrentLinkedQueue<LongPollingResource>();
private long _pollingInterval = 50;
private long _pollingLength = 10000;
private ProcessingRunnable _processingRunnable;
public LongPollingSystem() {
_waitingActions.add(new PauseResource());
}
public void start() {
_processingRunnable = new ProcessingRunnable();
Thread thr = new Thread(_processingRunnable);
thr.start();
}
public void appendLongPollingResource(LongPollingResource resource) {
_waitingActions.add(new TimeoutLongPollingResource(resource));
}
private void pause() {
try {
Thread.sleep(_pollingInterval);
} catch (InterruptedException exp) {
// Ignore
}
}
private class ProcessingRunnable implements Runnable {
@Override
public void run() {
while (true) {
LongPollingResource resource = _waitingActions.poll();
try {
if (resource.isChanged())
resource.process();
else
_waitingActions.add(resource);
} catch (Exception exp) {
_log.error("Error while processing a long-polled resource", exp);
}
}
}
}
private class TimeoutLongPollingResource implements LongPollingResource {
private long _start;
private LongPollingResource _longPollingResource;
private TimeoutLongPollingResource(LongPollingResource longPollingResource) {
_longPollingResource = longPollingResource;
_start = System.currentTimeMillis();
}
@Override
public boolean isChanged() {
boolean timeout = _start + _pollingLength < System.currentTimeMillis();
return timeout || _longPollingResource.isChanged();
}
@Override
public void process() {
_longPollingResource.process();
}
}
private class PauseResource implements LongPollingResource {
@Override
public boolean isChanged() {
return true;
}
@Override
public void process() {
pause();
_waitingActions.add(this);
}
}
}

3
gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LotroServerPipelineFactory.java
View File

@@ -29,6 +29,9 @@ public class LotroServerPipelineFactory implements ChannelPipelineFactory {
ServerBuilder.constructObjects(objects);
objects.put(LoggedUserHolder.class, new LoggedUserHolder());
LongPollingSystem longPollingSystem = new LongPollingSystem();
longPollingSystem.start();
objects.put(LongPollingSystem.class, longPollingSystem);
_context = objects;
_uriRequestHandler = new RootUriRequestHandler(_context);

72
gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/ChatRequestHandler.java
View File

@@ -3,6 +3,8 @@ package com.gempukku.lotro.async.handler;
import com.gempukku.lotro.PrivateInformationException;
import com.gempukku.lotro.SubscriptionExpiredException;
import com.gempukku.lotro.async.HttpProcessingException;
import com.gempukku.lotro.async.LongPollingResource;
import com.gempukku.lotro.async.LongPollingSystem;
import com.gempukku.lotro.async.ResponseWriter;
import com.gempukku.lotro.chat.ChatMessage;
import com.gempukku.lotro.chat.ChatRoomMediator;
@@ -29,10 +31,12 @@ public class ChatRequestHandler extends LotroServerRequestHandler implements Uri
private long _longPollingLength = 5000;
private long _longPollingInterval = 200;
private ChatServer _chatServer;
private LongPollingSystem _longPollingSystem;
public ChatRequestHandler(Map<Type, Object> context) {
super(context);
_chatServer = extractObject(context, ChatServer.class);
_longPollingSystem = extractObject(context, LongPollingSystem.class);
}
@Override
@@ -71,32 +75,56 @@ public class ChatRequestHandler extends LotroServerRequestHandler implements Uri
throw new HttpProcessingException(403);
}
try {
// Use long polling
long start = System.currentTimeMillis();
while (System.currentTimeMillis() < start + _longPollingLength && !chatRoom.hasPendingMessages(resourceOwner.getName())) {
try {
Thread.sleep(_longPollingInterval);
} catch (InterruptedException exp) {
ChatUpdateLongPollingResource polledResource = new ChatUpdateLongPollingResource(chatRoom, room, resourceOwner.getName(), responseWriter);
if (polledResource.isChanged())
polledResource.process();
else
_longPollingSystem.appendLongPollingResource(polledResource);
}
}
}
List<ChatMessage> chatMessages = chatRoom.getPendingMessages(resourceOwner.getName());
private class ChatUpdateLongPollingResource implements LongPollingResource {
private ChatRoomMediator _chatRoom;
private String _room;
private String _playerId;
private ResponseWriter _responseWriter;
Collection<String> usersInRoom = chatRoom.getUsersInRoom();
DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance();
DocumentBuilder documentBuilder = documentBuilderFactory.newDocumentBuilder();
Document doc = documentBuilder.newDocument();
serializeChatRoomData(room, chatMessages, usersInRoom, doc);
responseWriter.writeResponse(doc);
} catch (SubscriptionExpiredException exp) {
throw new HttpProcessingException(410);
private ChatUpdateLongPollingResource(ChatRoomMediator chatRoom, String room, String playerId, ResponseWriter responseWriter) {
_chatRoom = chatRoom;
_room = room;
_playerId = playerId;
_responseWriter = responseWriter;
}
@Override
public boolean isChanged() {
try {
return _chatRoom.hasPendingMessages(_playerId);
} catch (SubscriptionExpiredException e) {
return true;
}
}
@Override
public void process() {
try {
List<ChatMessage> chatMessages = _chatRoom.getPendingMessages(_playerId);
Collection<String> usersInRoom = _chatRoom.getUsersInRoom();
DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance();
DocumentBuilder documentBuilder = documentBuilderFactory.newDocumentBuilder();
Document doc = documentBuilder.newDocument();
serializeChatRoomData(_room, chatMessages, usersInRoom, doc);
_responseWriter.writeResponse(doc);
} catch (SubscriptionExpiredException exp) {
_responseWriter.writeError(410);
} catch (Exception exp) {
_responseWriter.writeError(500);
}
}
}
private void getMessages(HttpRequest request, String room, ResponseWriter responseWriter) throws Exception {

86
gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/HallRequestHandler.java
View File

@@ -3,6 +3,8 @@ package com.gempukku.lotro.async.handler;
import com.gempukku.lotro.SubscriptionConflictException;
import com.gempukku.lotro.SubscriptionExpiredException;
import com.gempukku.lotro.async.HttpProcessingException;
import com.gempukku.lotro.async.LongPollingResource;
import com.gempukku.lotro.async.LongPollingSystem;
import com.gempukku.lotro.async.ResponseWriter;
import com.gempukku.lotro.collection.CollectionsManager;
import com.gempukku.lotro.db.vo.League;
@@ -38,6 +40,7 @@ public class HallRequestHandler extends LotroServerRequestHandler implements Uri
private LotroCardBlueprintLibrary _library;
private long _longPollingLength = 5000;
private long _longPollingInterval = 200;
private LongPollingSystem _longPollingSystem;
public HallRequestHandler(Map<Type, Object> context) {
super(context);
@@ -46,6 +49,7 @@ public class HallRequestHandler extends LotroServerRequestHandler implements Uri
_hallServer = extractObject(context, HallServer.class);
_leagueService = extractObject(context, LeagueService.class);
_library = extractObject(context, LotroCardBlueprintLibrary.class);
_longPollingSystem = extractObject(context, LongPollingSystem.class);
}
@Override
@@ -305,33 +309,12 @@ public class HallRequestHandler extends LotroServerRequestHandler implements Uri
Player resourceOwner = getResourceOwnerSafely(request, participantId);
DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance();
DocumentBuilder documentBuilder = documentBuilderFactory.newDocumentBuilder();
Document doc = documentBuilder.newDocument();
Element hall = doc.createElement("hall");
try {
// Use long polling
long start = System.currentTimeMillis();
while (System.currentTimeMillis() < start + _longPollingLength && !_hallServer.hasChanges(resourceOwner, channelNumber)) {
try {
Thread.sleep(_longPollingInterval);
} catch (InterruptedException exp) {
}
}
_hallServer.processHall(resourceOwner, channelNumber, new SerializeHallInfoVisitor(doc, hall));
} catch (SubscriptionExpiredException exp) {
throw new HttpProcessingException(410);
} catch (SubscriptionConflictException exp) {
throw new HttpProcessingException(409);
}
hall.setAttribute("currency", String.valueOf(_collectionManager.getPlayerCollection(resourceOwner, "permanent").getCurrency()));
doc.appendChild(hall);
responseWriter.writeResponse(doc);
HallUpdateLongPollingResource polledResource = new HallUpdateLongPollingResource(resourceOwner, channelNumber, responseWriter);
if (polledResource.isChanged())
polledResource.process();
else
_longPollingSystem.appendLongPollingResource(polledResource);
} catch (HttpProcessingException exp) {
responseWriter.writeError(exp.getStatus());
} catch (Exception exp) {
@@ -339,6 +322,55 @@ public class HallRequestHandler extends LotroServerRequestHandler implements Uri
}
}
private class HallUpdateLongPollingResource implements LongPollingResource {
private Player _resourceOwner;
private int _channelNumber;
private ResponseWriter _responseWriter;
private HallUpdateLongPollingResource(Player resourceOwner, int channelNumber, ResponseWriter responseWriter) {
_resourceOwner = resourceOwner;
_channelNumber = channelNumber;
_responseWriter = responseWriter;
}
@Override
public boolean isChanged() {
try {
return _hallServer.hasChanges(_resourceOwner, _channelNumber);
} catch (SubscriptionExpiredException e) {
return true;
} catch (SubscriptionConflictException e) {
return true;
}
}
@Override
public void process() {
try {
DocumentBuilderFactory documentBuilderFactory = DocumentBuilderFactory.newInstance();
DocumentBuilder documentBuilder = documentBuilderFactory.newDocumentBuilder();
Document doc = documentBuilder.newDocument();
Element hall = doc.createElement("hall");
try {
_hallServer.processHall(_resourceOwner, _channelNumber, new SerializeHallInfoVisitor(doc, hall));
} catch (SubscriptionExpiredException exp) {
throw new HttpProcessingException(410);
} catch (SubscriptionConflictException exp) {
throw new HttpProcessingException(409);
}
hall.setAttribute("currency", String.valueOf(_collectionManager.getPlayerCollection(_resourceOwner, "permanent").getCurrency()));
doc.appendChild(hall);
_responseWriter.writeResponse(doc);
} catch (HttpProcessingException exp) {
_responseWriter.writeError(exp.getStatus());
} catch (Exception exp) {
_responseWriter.writeError(500);
}
}
}
private class SerializeHallInfoVisitor implements HallChannelVisitor {
private Document _doc;