diff --git a/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LongPollingResource.java b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LongPollingResource.java new file mode 100644 index 000000000..e84e77701 --- /dev/null +++ b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LongPollingResource.java @@ -0,0 +1,7 @@ +package com.gempukku.lotro.async; + +public interface LongPollingResource { + public boolean isChanged(); + + public void process(); +} diff --git a/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LongPollingSystem.java b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LongPollingSystem.java new file mode 100644 index 000000000..174d793ad --- /dev/null +++ b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LongPollingSystem.java @@ -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 _waitingActions = new ConcurrentLinkedQueue(); + 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); + } + } +} diff --git a/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LotroServerPipelineFactory.java b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LotroServerPipelineFactory.java index d5c1ab88c..760d41b30 100644 --- a/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LotroServerPipelineFactory.java +++ b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/LotroServerPipelineFactory.java @@ -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); diff --git a/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/ChatRequestHandler.java b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/ChatRequestHandler.java index 36cceaf5b..147298617 100644 --- a/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/ChatRequestHandler.java +++ b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/ChatRequestHandler.java @@ -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 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 chatMessages = chatRoom.getPendingMessages(resourceOwner.getName()); + private class ChatUpdateLongPollingResource implements LongPollingResource { + private ChatRoomMediator _chatRoom; + private String _room; + private String _playerId; + private ResponseWriter _responseWriter; - Collection 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 chatMessages = _chatRoom.getPendingMessages(_playerId); + + Collection 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 { diff --git a/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/HallRequestHandler.java b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/HallRequestHandler.java index fdd82da4c..2bca41d00 100644 --- a/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/HallRequestHandler.java +++ b/gemp-lotr/gemp-lotr-async/src/main/java/com/gempukku/lotro/async/handler/HallRequestHandler.java @@ -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 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;