[32] | 1 | package geniusweb.partiesserver.websocket;
|
---|
| 2 |
|
---|
| 3 | import java.io.EOFException;
|
---|
| 4 | import java.io.IOException;
|
---|
| 5 | import java.net.URI;
|
---|
| 6 | import java.util.Date;
|
---|
| 7 | import java.util.List;
|
---|
| 8 | import java.util.concurrent.CopyOnWriteArrayList;
|
---|
| 9 | import java.util.logging.Level;
|
---|
| 10 |
|
---|
| 11 | import javax.websocket.CloseReason;
|
---|
| 12 | import javax.websocket.CloseReason.CloseCodes;
|
---|
| 13 | import javax.websocket.OnClose;
|
---|
| 14 | import javax.websocket.OnError;
|
---|
| 15 | import javax.websocket.OnMessage;
|
---|
| 16 | import javax.websocket.OnOpen;
|
---|
| 17 | import javax.websocket.Session;
|
---|
| 18 | import javax.websocket.server.PathParam;
|
---|
| 19 | import javax.websocket.server.ServerEndpoint;
|
---|
| 20 |
|
---|
| 21 | import geniusweb.actions.Action;
|
---|
| 22 | import geniusweb.actions.PartyId;
|
---|
| 23 | import geniusweb.connection.ConnectionEnd;
|
---|
| 24 | import geniusweb.inform.Inform;
|
---|
| 25 | import geniusweb.inform.Settings;
|
---|
| 26 | import geniusweb.partiesserver.Jackson;
|
---|
| 27 | import geniusweb.partiesserver.repository.RunningPartiesRepo;
|
---|
| 28 | import geniusweb.partiesserver.repository.RunningParty;
|
---|
| 29 | import geniusweb.party.Party;
|
---|
| 30 | import geniusweb.references.Reference;
|
---|
| 31 | import tudelft.utilities.listener.Listener;
|
---|
| 32 | import tudelft.utilities.logging.ReportToLogger;
|
---|
| 33 | import tudelft.utilities.logging.Reporter;
|
---|
| 34 |
|
---|
| 35 | /**
|
---|
| 36 | * Returns a websocket that communicates with the {@link RunningParty}. The
|
---|
| 37 | * websocket is initiated/called/requested by an external party, typicall hte
|
---|
| 38 | * protocol running on another server. We close the socket when we detect the
|
---|
| 39 | * party is gone (eg, due to a bug in the party or a time-out). If the socket
|
---|
| 40 | * breaks, we close the party as it becomes isolated.
|
---|
| 41 | */
|
---|
| 42 | @ServerEndpoint("/party/{party}")
|
---|
| 43 | public class PartySocket implements ConnectionEnd<Inform, Action> {
|
---|
| 44 | private static final int MAX_MESSAGE_SIZE = 20 * 1024 * 1024;
|
---|
| 45 |
|
---|
| 46 | private final Reporter log;
|
---|
| 47 | private final RunningPartiesRepo runningparties;
|
---|
| 48 | private final List<Listener<Inform>> listeners = new CopyOnWriteArrayList<Listener<Inform>>();
|
---|
| 49 | private final ActiveThreads threads;
|
---|
| 50 |
|
---|
| 51 | // should all be final, except that we can only set them when start is
|
---|
| 52 | // called...
|
---|
| 53 | private Session session;
|
---|
| 54 |
|
---|
| 55 | private PartyId partyID;
|
---|
| 56 |
|
---|
| 57 | private SendBuffer outstream;
|
---|
| 58 | private Throwable error = null;
|
---|
| 59 |
|
---|
| 60 | public PartySocket() {
|
---|
| 61 | this(RunningPartiesRepo.instance(),
|
---|
| 62 | new ReportToLogger("partiesserver"));
|
---|
| 63 | }
|
---|
| 64 |
|
---|
| 65 | public PartySocket(RunningPartiesRepo parties, Reporter reporter) {
|
---|
| 66 | this.runningparties = parties;
|
---|
| 67 | this.log = reporter;
|
---|
| 68 | this.threads = new ActiveThreads(log);
|
---|
| 69 | }
|
---|
| 70 |
|
---|
| 71 | @OnOpen
|
---|
| 72 | public void start(Session session,
|
---|
| 73 | @PathParam("party") final String partyidname) throws IOException {
|
---|
| 74 | if (partyidname == null) {
|
---|
| 75 | throw new IllegalArgumentException("party can't be null");
|
---|
| 76 | }
|
---|
| 77 | session.setMaxTextMessageBufferSize(MAX_MESSAGE_SIZE);
|
---|
| 78 | session.setMaxBinaryMessageBufferSize(MAX_MESSAGE_SIZE);
|
---|
| 79 |
|
---|
| 80 | this.session = session;
|
---|
| 81 | this.partyID = new PartyId(partyidname);
|
---|
| 82 | this.outstream = new SendBuffer(session.getBasicRemote(), log);
|
---|
| 83 |
|
---|
| 84 | // listen to parties repo, so that we can act if party is terminated
|
---|
| 85 | runningparties.addListener(new Listener<PartyId>() {
|
---|
| 86 | @Override
|
---|
| 87 | public void notifyChange(PartyId data) {
|
---|
| 88 | if (runningparties.get(partyID) == null) {
|
---|
| 89 | // party was removed, probably due to time out.
|
---|
| 90 | try {
|
---|
| 91 | log.log(Level.INFO,
|
---|
| 92 | "Party " + partyID + " was terminated ");
|
---|
| 93 | runningparties.removeListener(this);
|
---|
| 94 | session.close(new CloseReason(CloseCodes.GOING_AWAY,
|
---|
| 95 | "detected that party vanished"));
|
---|
| 96 | } catch (IOException e) {
|
---|
| 97 | log.log(Level.WARNING,
|
---|
| 98 | "failed to close the socket for " + partyID, e);
|
---|
| 99 | }
|
---|
| 100 | }
|
---|
| 101 | }
|
---|
| 102 | });
|
---|
| 103 | RunningParty runningparty = runningparties.get(partyID);
|
---|
| 104 |
|
---|
| 105 | if (runningparty == null) {
|
---|
| 106 | session.close(new CloseReason(CloseReason.CloseCodes.CANNOT_ACCEPT,
|
---|
| 107 | "No such party: " + partyID));
|
---|
| 108 | } else {
|
---|
| 109 | runningparties.replace(runningparty.withConnection(this));
|
---|
| 110 | }
|
---|
| 111 |
|
---|
| 112 | // we do not listen for changes on the Parties factory as we can't
|
---|
| 113 | // replace the running party anyway
|
---|
| 114 | }
|
---|
| 115 |
|
---|
| 116 | /**
|
---|
| 117 | * Called when a message comes in on the server socket that needs to be
|
---|
| 118 | * passed into the {@link Party}. We also sniff the message for termination
|
---|
| 119 | * indicators.
|
---|
| 120 | *
|
---|
| 121 | * @param informmessage the incoming string, supposedly a JSON-fornatted
|
---|
| 122 | * {@link Inform}
|
---|
| 123 | * @param session the session. We already have this info.
|
---|
| 124 | * @throws IOException if there is a socket/connection error, we can't parse
|
---|
| 125 | * the received data, etc
|
---|
| 126 | */
|
---|
| 127 | @OnMessage
|
---|
| 128 | public void onMessage(String informmessage, Session session)
|
---|
| 129 | throws IOException {
|
---|
| 130 | if (this.session != session) {
|
---|
| 131 | throw new IllegalStateException("Unexpected change of session ID");
|
---|
| 132 | }
|
---|
| 133 | if (!threads.isEmpty()) {
|
---|
| 134 | log.log(Level.WARNING, "Party " + partyID
|
---|
| 135 | + " is still busy with previous message");
|
---|
| 136 | }
|
---|
| 137 | Inform info = Jackson.instance().readValue(informmessage, Inform.class);
|
---|
| 138 | RunningParty party = runningparties.get(partyID);
|
---|
| 139 | if (party != null) {
|
---|
| 140 | // first sniff, to ensure the deadlines are updated first.
|
---|
| 141 | sniff(info, party);
|
---|
| 142 | log.log(Level.FINE, "Inform " + partyID + ": " + info);
|
---|
| 143 | try {
|
---|
| 144 | threads.add();
|
---|
| 145 | party.inform(info);
|
---|
| 146 | threads.remove();
|
---|
| 147 | } catch (Throwable e) {
|
---|
| 148 | threads.remove();
|
---|
| 149 | if (e instanceof ThreadDeath)
|
---|
| 150 | /*
|
---|
| 151 | * ThreadDeath always prints stacktrace anwyay, stacktrace
|
---|
| 152 | * is useless because it contains the thrower's stacktrace
|
---|
| 153 | * instead of a useful message.
|
---|
| 154 | */
|
---|
| 155 | log.log(Level.WARNING,
|
---|
| 156 | "Party was killed while handling inform.");
|
---|
| 157 | else
|
---|
| 158 | log.log(Level.WARNING, "Party failed on inform.", e);
|
---|
| 159 | /*
|
---|
| 160 | * severe as someone wants to debug that. Not severe for us we
|
---|
| 161 | * don't use jackson.writeValueAsString(e) here because
|
---|
| 162 | * CloseReason has 123 char limit. The error will be too large
|
---|
| 163 | * to fit and will be truncated therefore we send a plain string
|
---|
| 164 | * and hope that enough will arrive at the other side (the
|
---|
| 165 | * protocol)
|
---|
| 166 | */
|
---|
| 167 | session.close(new CloseReason(
|
---|
| 168 | CloseReason.CloseCodes.CLOSED_ABNORMALLY,
|
---|
| 169 | "party threw exception: " + collectReasons(e)));
|
---|
| 170 | }
|
---|
| 171 | } // else party is dead but that's handled in the listener above
|
---|
| 172 | }
|
---|
| 173 |
|
---|
| 174 | /**
|
---|
| 175 | *
|
---|
| 176 | * @param e an exception
|
---|
| 177 | * @return collect all reasons and sub-reasons
|
---|
| 178 | */
|
---|
| 179 | private String collectReasons(Throwable e) {
|
---|
| 180 | String reason = "";
|
---|
| 181 | do {
|
---|
| 182 | reason = reason + " : " + e.getClass().getSimpleName() + ":"
|
---|
| 183 | + e.getMessage();
|
---|
| 184 | e = e.getCause();
|
---|
| 185 | } while (e != null);
|
---|
| 186 | return reason;
|
---|
| 187 | }
|
---|
| 188 |
|
---|
| 189 | /**
|
---|
| 190 | * Sniffs the Inform for termination-related information
|
---|
| 191 | *
|
---|
| 192 | * @param info
|
---|
| 193 | * @param party the party we're handling
|
---|
| 194 | * @throws IOException
|
---|
| 195 | */
|
---|
| 196 | private void sniff(Inform info, RunningParty party) throws IOException {
|
---|
| 197 | // do NOT sniff Finished messages and close sockets. Parties and their
|
---|
| 198 | // sockets need to stay running till all messages are handled
|
---|
| 199 | if (info instanceof Settings) {
|
---|
| 200 | Date end = ((Settings) info).getProgress().getTerminationTime();
|
---|
| 201 | runningparties.replace(party.withEndTime(end));
|
---|
| 202 | }
|
---|
| 203 | }
|
---|
| 204 |
|
---|
| 205 | @OnClose
|
---|
| 206 | public void onClose() throws IOException {
|
---|
| 207 | log.log(Level.INFO, "socket closed to " + partyID);
|
---|
| 208 | if (!threads.isEmpty()) {
|
---|
| 209 | log.log(Level.WARNING, "Party " + partyID
|
---|
| 210 | + " failed to terminate. Trying to kill the remaining threads.");
|
---|
| 211 | threads.killall();
|
---|
| 212 | }
|
---|
| 213 | runningparties.remove(partyID);
|
---|
| 214 | outstream.stop();
|
---|
| 215 | }
|
---|
| 216 |
|
---|
| 217 | @OnError
|
---|
| 218 | public void onError(Throwable t) throws Throwable {
|
---|
| 219 | if (t instanceof EOFException) {
|
---|
| 220 | // This is a standard exception from Apache when a socket closes.
|
---|
| 221 | // Just ignore
|
---|
| 222 | // it...
|
---|
| 223 | log.log(Level.FINEST,
|
---|
| 224 | "apache reported EOF from (probably closed) socket, ignoring");
|
---|
| 225 | return;
|
---|
| 226 | }
|
---|
| 227 | log.log(Level.WARNING, "Unhandled exception was reported by Tomcat", t);
|
---|
| 228 | }
|
---|
| 229 |
|
---|
| 230 | @Override
|
---|
| 231 | public void send(Action action) throws IOException {
|
---|
| 232 | if (action == null) {
|
---|
| 233 | log.log(Level.INFO, "null action received. Closing websocket");
|
---|
| 234 | // signals that party closed the stream
|
---|
| 235 | close();
|
---|
| 236 | return;
|
---|
| 237 | }
|
---|
| 238 | log.log(Level.FINE, "Action " + partyID + ": " + action);
|
---|
| 239 | outstream.send(Jackson.instance().writeValueAsString(action));
|
---|
| 240 | }
|
---|
| 241 |
|
---|
| 242 | @Override
|
---|
| 243 | public Reference getReference() {
|
---|
| 244 | return null;
|
---|
| 245 | }
|
---|
| 246 |
|
---|
| 247 | @Override
|
---|
| 248 | public URI getRemoteURI() {
|
---|
| 249 | // In serverendpoints we apparently can't see who was contacting us.
|
---|
| 250 | throw new UnsupportedOperationException();
|
---|
| 251 | }
|
---|
| 252 |
|
---|
| 253 | @Override
|
---|
| 254 | public void close() {
|
---|
| 255 | try {
|
---|
| 256 | session.close();
|
---|
| 257 | } catch (IOException e) {
|
---|
| 258 | log.log(Level.SEVERE, "failed to close " + partyID, e);
|
---|
| 259 | }
|
---|
| 260 | }
|
---|
| 261 |
|
---|
| 262 | @Override
|
---|
| 263 | public Throwable getError() {
|
---|
| 264 | return error;
|
---|
| 265 | }
|
---|
| 266 |
|
---|
| 267 | @Override
|
---|
| 268 | public void addListener(Listener<Inform> l) {
|
---|
| 269 | listeners.add(l);
|
---|
| 270 | }
|
---|
| 271 |
|
---|
| 272 | @Override
|
---|
| 273 | public void removeListener(Listener<Inform> l) {
|
---|
| 274 | listeners.remove(l);
|
---|
| 275 | }
|
---|
| 276 |
|
---|
| 277 | /**
|
---|
| 278 | * This should only be called by the owner of the listenable, not by
|
---|
| 279 | * listeners or others. Avoid calling this from synchronized blocks as a
|
---|
| 280 | * notified listener might immediately make more calls to you.
|
---|
| 281 | * <p>
|
---|
| 282 | * Any listeners that throw an exception will be intercepted and their
|
---|
| 283 | * stacktrace is printed.
|
---|
| 284 | *
|
---|
| 285 | * @param data information about the change.
|
---|
| 286 | */
|
---|
| 287 | public void notifyListeners(Inform data) {
|
---|
| 288 | for (Listener<Inform> l : listeners) {
|
---|
| 289 | l.notifyChange(data);
|
---|
| 290 | }
|
---|
| 291 | }
|
---|
| 292 |
|
---|
| 293 | }
|
---|