import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
    Enumeration<String> acceptValues = request.getHeaders("Accept");
    while (acceptValues.hasMoreElements()) {
        String accept = acceptValues.nextElement();
        if (accept.equals("text/event-stream")) {
            EventSource eventSource = newEventSource(request);
            if (eventSource == null) {
                response.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
            } else {
                respond(request, response);
                Continuation continuation = ContinuationSupport.getContinuation(request);
                // Infinite timeout because the continuation is never resumed,
                // but only completed on close
                continuation.setTimeout(0L);
                continuation.suspend(response);
                EventSourceEmitter emitter = new EventSourceEmitter(eventSource, continuation);
                emitter.scheduleHeartBeat();
                open(eventSource, emitter);
            }
            return;
        }
    }
    super.doGet(request, response);
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
  metrics.getRequestsMeter().mark();

  if (!target.equals(s3Configuration.getLocalDownloadPath())) {
    metrics.getClientErrorsMeter().mark();
    response.sendError(404);
    return;
  }

  if (!request.getMethod().equalsIgnoreCase(HttpMethod.POST.name())) {
    metrics.getClientErrorsMeter().mark();
    response.sendError(405);
    return;
  }

  Optional<ArtifactDownloadRequest> artifactOptional = readDownloadRequest(request);

  if (!artifactOptional.isPresent()) {
    metrics.getClientErrorsMeter().mark();
    response.sendError(400);
    return;
  }

  Continuation continuation = ContinuationSupport.getContinuation(request);
  continuation.suspend(response);

  downloaderCoordinator.register(continuation, artifactOptional.get());
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
@Override
protected void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException {
  Continuation continuation = ContinuationSupport.getContinuation(request);
  if (continuation.isExpired()) {
    // timeout - just send a blank response and tell the connection that its continuation has timed out
    String clientId = (String) continuation.getAttribute(CLIENT_ID);
    if (clientId != null) {
      // TODO will this always get the correct continuation?
      _connectionManager.longPollHttpTimeout(clientId, continuation);
    }
    return;
  }
  String results = (String) request.getAttribute(RESULTS);
  // if this is the first time the request has been dispatched the results will be null. if the request has been
  // dispatched before and is being dispatched again after its continuation was resumed the results will be populated
  if (results == null) {
    setUpConnection(continuation, request, response);
  } else {
    // Send the results
    s_logger.debug("Writing results to HTTP response {}", results);
    response.getWriter().write(results);
  }
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
/**
 * Invoked when a client establishes a long-polling HTTP connection.
 * @param continuation The connection's continuation
 */
/* package */ void connect(Continuation continuation) {
  synchronized (_lock) {
    s_logger.debug("Long polling connection established, resetting timeout task {}", _timeoutTask);
    _timeoutTask.reset();
    _continuation = continuation;
    _continuation.setTimeout(10000);
    // if there are updates queued sent them immediately otherwise save the continuation until an update
    if (!_updates.isEmpty()) {
      try {
        sendUpdate(formatUpdate(_updates));
      } catch (JSONException e) {
        // this shouldn't ever happen, the updates are all URLs
        s_logger.warn("Unable to format updates as JSON. updates: " + _updates, e);
      }
      _updates.clear();
    }
  }
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
@Override
protected void doGet(HttpServletRequest req, HttpServletResponse resp)
    throws ServletException, IOException {

  logger.info("doGet:" + Thread.currentThread().getName());
  final Continuation cnt = ContinuationSupport.getContinuation(req);

  //aka retry
  //cnt.setTimeout(5000);
  cnt.suspend(resp);

  webService.htmlStream().subscribe(new Action1<String>() {
    @Override
    public void call(String response) {
      logger.info("Notify:" + Thread.currentThread().getName());
      respond(response + "\n", cnt);
    }
  }, new Action1<Throwable>() {
    @Override
    public void call(Throwable throwable) {
      respond(throwable.getMessage(), cnt);
    }
  });
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
public void flushUpdates() {
	long idThread = Thread.currentThread().getId();
	
	List<Integer> players = pendingWakeUps.get(idThread);
	
	if (players != null) {
		synchronized (players) {
			for (Integer idPlayer : players) {
				Player player = DataAccess.getPlayerById(idPlayer);
				Continuation continuation = player.getContinuation();
				if (continuation != null)
					continuation.resume();
			}
		}
		players.clear();
	}
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
public SingularityS3DownloaderAsyncHandler(ArtifactManager artifactManager, ArtifactDownloadRequest artifactDownloadRequest, Continuation continuation, SingularityS3DownloaderMetrics metrics, SingularityRunnerExceptionNotifier exceptionNotifier) {
  this.artifactManager = artifactManager;
  this.artifactDownloadRequest = artifactDownloadRequest;
  this.continuation = continuation;
  this.metrics = metrics;
  this.start = System.currentTimeMillis();
  this.exceptionNotifier = exceptionNotifier;
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
private void setUpConnection(Continuation continuation, HttpServletRequest request, HttpServletResponse response) throws IOException {
  // suspend the request
  continuation.suspend(); // always suspend before registration
  String userName = (AuthUtils.isPermissive() ? null : AuthUtils.getUserName());
  // get the client ID from the URL and pass the continuation to the connection manager for the next updates
  String clientId = getClientId(request);
  boolean connected = (clientId != null) && _connectionManager.longPollHttpConnect(userName, clientId, continuation);
  if (!connected) {
    // couldn't get the client ID from the URL or the client ID didn't correspond to a known client
    // TODO how do I send something other than jetty's standard HTML error page?
    response.sendError(404, "Problem accessing " + request.getRequestURI() + ".  Reason: Unknown client ID " + clientId);
    continuation.complete();
  }
  continuation.setAttribute(CLIENT_ID, clientId);
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
/**
 * Associates a continuation with a client connection so asynchronous updates can be pushed to the client.
 * 
 * @param userId The ID of the user
 * @param clientId The client ID of the connection
 * @param continuation For sending an async response to the client
 * @return true if the connection was successful, false if the client ID doesn't correspond to
 * an existing connection
 */
/* package */ boolean longPollHttpConnect(String userId, String clientId, Continuation continuation) {
  // TODO check args
  LongPollingUpdateListener listener = _updateListeners.get(clientId);
  if (listener != null) {
    if (!Objects.equal(userId, listener.getUserId())) {
      throw new IllegalArgumentException("User ID " + userId + " doesn't correspond to client ID: " + clientId);
    }
    listener.connect(continuation);
    return true;
  } else {
    return false;
  }
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
public JSONProcCallback(Request request, Continuation continuation, String jsonp) {
    assert(request != null);
    assert(continuation != null);

    m_request = request;
    m_continuation = continuation;
    m_jsonp = jsonp;
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
@Override
public void handleAsynchronous(Request baseRequest,
    HttpServletRequest request, HttpServletResponse response) {
  // Jetty continuations
  // TODO(spenceg) Try Servlet 3.0 if this doesn't work
  Continuation continuation = ContinuationSupport.getContinuation(request);
  continuation.suspend(response); //Start Async Processing

  TranslationRequest translationRequest = (TranslationRequest) baseRequest;
  
  // Translate to uppercase!
  List<String> translation = Arrays.asList(translationRequest.text.toUpperCase().split("\\s+"));
  List<String> alignments = new ArrayList<>(2);
  alignments.add("1-2");
  alignments.add("2-1");
  TranslationQuery query = new TranslationQuery(translation, alignments, 1.0);
  List<TranslationQuery> queryList = new ArrayList<>(1);
  queryList.add(query);
  Type t = new TypeToken<TranslationReply>() {}.getType();
  TranslationReply baseResponse = new TranslationReply(queryList);

  // Simulate a long call to the MT system
  Random random = new Random();
  try {
    Thread.sleep(500 + random.nextInt(1000));
  } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
  }
  
  ServiceResponse serviceResponse = new ServiceResponse(baseResponse, t);
  request.setAttribute(PhrasalServlet.ASYNC_KEY, serviceResponse);   
  continuation.resume(); // Re-dispatch/ resume to generate response
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
public DecoderInput(int inputId, String text, String prefix, int n, Language targetLanguage, String inputProps, HttpServletRequest request,
    Continuation continuation) {
  this.inputId = inputId;
  this.text = text;
  this.tgtPrefix = prefix;
  this.properties = InputProperties.fromString(inputProps);
  this.targetLanguage = targetLanguage;
  this.n = n;
  this.request = request;
  this.continuation = continuation;
  this.submitTime = System.nanoTime();
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
public void getOneTimeGeoLocation(Continuation cont, Request req) {
    this.continuation = cont;
    this.baseRequest = req;
    final Context app = BreadApp.getBreadContext();
    if (app == null)
        return;
    locationManager = (LocationManager) app.getSystemService(Context.LOCATION_SERVICE);
    if (locationManager == null) {
        Log.e(TAG, "getOneTimeGeoLocation: locationManager is null!");
        return;
    }
    BRExecutor.getInstance().forMainThreadTasks().execute(new Runnable() {
        @Override
        public void run() {
            if (ActivityCompat.checkSelfPermission(app, Manifest.permission.ACCESS_FINE_LOCATION) !=
                    PackageManager.PERMISSION_GRANTED && ActivityCompat.checkSelfPermission(app, Manifest.permission.ACCESS_COARSE_LOCATION) != PackageManager.PERMISSION_GRANTED) {
                RuntimeException ex = new RuntimeException("getOneTimeGeoLocation, can't happen");
                Log.e(TAG, "run: getOneTimeGeoLocation, can't happen");
                BRReportsManager.reportBug(ex);
                return;
            }
            locationManager.requestLocationUpdates(LocationManager.NETWORK_PROVIDER, 0, 0, locationListener);
            locationManager.requestLocationUpdates(LocationManager.GPS_PROVIDER, 0, 0, locationListener);
        }
    });

}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
private void respond(String ctx, Continuation cnt) {
  try {
    final Writer wr = cnt.getServletResponse().getWriter();
    wr.write(ctx);
    wr.flush();
    wr.close();
  } catch (IOException e) {
  } finally {
    cnt.complete();
  }
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
private void handlePackets(Request baseRequest, HttpServletRequest request,
		HttpServletResponse response) throws IOException {
	String domain = request.getParameter("domain");
	Ticket ticket = Ticket.parse(request.getParameter("ticket"));
	EndOid useroid = makeoid(domain, ticket);
	String callback = g(request, "callback");

	Subscriber subscriber = router.lookup(useroid, ticket);
	if (subscriber == null) {
		jsonReturn(
				new JSONResult(
						callback
								+ "({\"status\": \"error\", \"message\": \"subscriber not found\"})"),
				response, HttpServletResponse.SC_BAD_REQUEST);
		return;
	}
	synchronized (subscriber) {
		if (subscriber.queue.size() > 0) {
			List<Packet> packets = new ArrayList<Packet>();
			Packet packet = null;
			while ((packet = subscriber.queue.poll()) != null) {
				packets.add(packet);
			}
			JSONResult result = new Encoder(packets).jsonResult();
			// Send one chat message
			response.setContentType("application/javascript;charset=utf-8");
			response.setStatus(HttpServletResponse.SC_OK);
			response.getOutputStream().write((callback + "(").getBytes());
			response.getOutputStream().write(result.toJSON());
			response.getOutputStream().write(new byte[] { ')' });
			response.getOutputStream().flush();
		} else {
			Continuation continuation = ContinuationSupport
					.getContinuation(request);
			if (continuation.isInitial()) {
				// No chat in queue, so suspend and wait for timeout or chat
				continuation.setTimeout(20000);
				continuation.suspend();
				subscriber.continuation = continuation;
			} else {
				// Timeout so send empty response
				response.setContentType("application/javascript;charset=utf-8");
				response.setStatus(HttpServletResponse.SC_OK);
				response.getWriter().print(
						callback + "({\"status\": \"ok\"})");
				response.getWriter().flush();
			}
		}
		subscriber.lastActive = System.currentTimeMillis();

	}
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
@Test
public void aliasMiss() throws Exception {
    AliasHandler handler = new AliasHandler(commandProcessor, true, pathAliases);
    Continuation continuation = mock(Continuation.class);
    when(request.getAttribute(Continuation.ATTRIBUTE)).thenReturn(continuation);
    handler.handle("/missPath", baseRequest, request, response);
    verify(commandProcessor, times(1)).process(any(HttpCommand.class));
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException {
  metrics.getRequestsMeter().mark();

  if (!target.equals(s3Configuration.getLocalDownloadPath())) {
    metrics.getClientErrorsMeter().mark();
    response.sendError(404);
    return;
  }

  if (!request.getMethod().equalsIgnoreCase(HttpMethod.POST.name())) {
    metrics.getClientErrorsMeter().mark();
    response.sendError(405);
    return;
  }

  Optional<ArtifactDownloadRequest> artifactOptional = readDownloadRequest(request);

  if (!artifactOptional.isPresent()) {
    metrics.getClientErrorsMeter().mark();
    response.sendError(400);
    return;
  }

  Continuation continuation = ContinuationSupport.getContinuation(request);
  continuation.suspend(response);
  if (artifactOptional.get().getTimeoutMillis().isPresent()) {
    continuation.setTimeout(artifactOptional.get().getTimeoutMillis().get());
  }

  downloaderCoordinator.register(continuation, artifactOptional.get());
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
public SingularityS3DownloaderAsyncHandler(ArtifactManager artifactManager, ArtifactDownloadRequest artifactDownloadRequest, Continuation continuation, SingularityS3DownloaderMetrics metrics,
    SingularityRunnerExceptionNotifier exceptionNotifier, DownloadListener downloadListener) {
  this.artifactManager = artifactManager;
  this.artifactDownloadRequest = artifactDownloadRequest;
  this.continuation = continuation;
  this.metrics = metrics;
  this.start = System.currentTimeMillis();
  this.exceptionNotifier = exceptionNotifier;
  this.downloadListener = downloadListener;
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
public void setContinuation(Continuation continuation) {
	Player cachedPlayer = DataAccess.getPlayerById(getId());
	
	if (cachedPlayer != this)
		cachedPlayer.continuation = continuation;
	this.continuation = continuation;
}
 

import org.eclipse.jetty.continuation.Continuation; //导入依赖的package包/类
public EventSourceEmitter(EventSource eventSource, Continuation continuation) throws IOException {
    this.eventSource = eventSource;
    this.continuation = continuation;
    this.output = continuation.getServletResponse().getOutputStream();
}