/*** Core Zuul servlet which intializes and orchestrates zuulFilter execution** @author Mikey Cohen*         Date: 12/23/11*         Time: 10:44 AM*/@Overridepublic void service(javax.servlet.ServletRequest servletRequest, javax.servlet.ServletResponse servletResponse) throws ServletException, IOException {try {init((HttpServletRequest) servletRequest, (HttpServletResponse) servletResponse);// Marks this request as having passed through the "Zuul engine", as opposed to servlets// explicitly bound in web.xml, for which requests will not have the same data attachedRequestContext context = RequestContext.getCurrentContext();context.setZuulEngineRan();try {preRoute();} catch (ZuulException e) {error(e);postRoute();return;}try {route();} catch (ZuulException e) {error(e);postRoute();return;}try {postRoute();} catch (ZuulException e) {error(e);return;}} catch (Throwable e) {error(new ZuulException(e, 500, "UNHANDLED_EXCEPTION_" + e.getClass().getName()));} finally {RequestContext.getCurrentContext().unset();}}

/*** This class manages the directory polling for changes and new Groovy filters.* Polling interval and directories are specified in the initialization of the class, and a poller will check* for changes and additions.** @author Mikey Cohen*         Date: 12/7/11*         Time: 12:09 PM*/void processGroovyFiles(List<File> aFiles) throws Exception {List<Callable<Boolean>> tasks = new ArrayList<>();for (File file : aFiles) {tasks.add(() -> {try {return filterLoader.putFilter(file);}catch(Exception e) {LOG.error("Error loading groovy filter from disk! file = " + String.valueOf(file), e);return false;}});}processFilesService.invokeAll(tasks, FILE_PROCESSOR_TASKS_TIMEOUT_SECS.get(), TimeUnit.SECONDS);}

/*** Servlet for uploading/downloading/managing scripts.* <p/>* <ul>* <li>Upload scripts to the registry for a given endpoint.</li>* <li>Download scripts from the registry</li>* <li>List all revisions of scripts for a given endpoint.</li>* <li>Mark a particular script revision as active for production.</li>* </ul>*/@Overrideprotected void doPut(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {if (!adminEnabled.get()) {response.sendError(HttpServletResponse.SC_FORBIDDEN, "Filter admin is disabled. See the zuul.filters.admin.enabled FastProperty.");return;}// retrieve arguments and validateString action = request.getParameter("action");/* validate the action and method */if (!isValidAction(request, response)) {return;}// perform actionif ("UPLOAD".equals(action)) {handleUploadAction(request, response);} else if ("ACTIVATE".equals(action)) {handleActivateAction(request, response);} else if ("CANARY".equals(action)) {handleCanaryAction(request, response);} else if ("DEACTIVATE".equals(action)) {handledeActivateAction(request, response);}}

    @Overrideprotected void initChannel(Channel ch) throws Exception{// Configure our pipeline of ChannelHandlerS.ChannelPipeline pipeline = ch.pipeline();storeChannel(ch);addTimeoutHandlers(pipeline);addPassportHandler(pipeline);addTcpRelatedHandlers(pipeline);addHttp1Handlers(pipeline);addHttpRelatedHandlers(pipeline);addZuulHandlers(pipeline);}

   protected void addZuulHandlers(final ChannelPipeline pipeline){pipeline.addLast("logger", nettyLogger);pipeline.addLast(new ClientRequestReceiver(sessionContextDecorator));pipeline.addLast(passportLoggingHandler);addZuulFilterChainHandler(pipeline);pipeline.addLast(new ClientResponseWriter(requestCompleteHandler, registry));}protected void addZuulFilterChainHandler(final ChannelPipeline pipeline) {final ZuulFilter<HttpResponseMessage, HttpResponseMessage>[] responseFilters = getFilters(   //1new OutboundPassportStampingFilter(FILTERS_OUTBOUND_START),new OutboundPassportStampingFilter(FILTERS_OUTBOUND_END));// response filter chainfinal ZuulFilterChainRunner<HttpResponseMessage> responseFilterChain = getFilterChainRunner(responseFilters,filterUsageNotifier);// endpoint | response filter chain
        final FilterRunner<HttpRequestMessage, HttpResponseMessage> endPoint = getEndpointRunner(responseFilterChain,  //2filterUsageNotifier, filterLoader);final ZuulFilter<HttpRequestMessage, HttpRequestMessage>[] requestFilters = getFilters(                   //3new InboundPassportStampingFilter(FILTERS_INBOUND_START),new InboundPassportStampingFilter(FILTERS_INBOUND_END));// request filter chain | end point | response filter chainfinal ZuulFilterChainRunner<HttpRequestMessage> requestFilterChain = getFilterChainRunner(requestFilters,filterUsageNotifier, endPoint);pipeline.addLast(new ZuulFilterChainHandler(requestFilterChain, responseFilterChain));}

    @Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof HttpRequestMessage) {zuulRequest = (HttpRequestMessage)msg;//Replace NETTY_SERVER_CHANNEL_HANDLER_CONTEXT in SessionContextfinal SessionContext zuulCtx = zuulRequest.getContext();zuulCtx.put(NETTY_SERVER_CHANNEL_HANDLER_CONTEXT, ctx);zuulCtx.put(ZUUL_FILTER_CHAIN, requestFilterChain);requestFilterChain.filter(zuulRequest);}else if ((msg instanceof HttpContent)&&(zuulRequest != null)) {requestFilterChain.filter(zuulRequest, (HttpContent) msg);}else {LOG.debug("Received unrecognized message type. " + msg.getClass().getName());ReferenceCountUtil.release(msg);}}

 @Overridepublic void filter(T inMesg, HttpContent chunk) {String filterName = "-";try {Preconditions.checkNotNull(inMesg, "input message");final AtomicInteger runningFilterIdx = getRunningFilterIndex(inMesg);final int limit = runningFilterIdx.get();for (int i = 0; i < limit; i++) {final ZuulFilter<T, T> filter = filters[i];filterName = filter.filterName();if ((! filter.isDisabled()) && (! shouldSkipFilter(inMesg, filter))) {final HttpContent newChunk = filter.processContentChunk(inMesg, chunk);if (newChunk == null)  {//Filter wants to break the chain and stop propagating this chunk any furtherreturn;}//deallocate original chunk if necessaryif ((newChunk != chunk) && (chunk.refCnt() > 0)) {chunk.release(chunk.refCnt());}chunk = newChunk;}}if (limit >= filters.length) {//Filter chain has run to end, pass down the channel pipeline
                invokeNextStage(inMesg, chunk);} else {inMesg.bufferBodyContents(chunk);boolean isAwaitingBody = isFilterAwaitingBody(inMesg);// Record passport states for start and end of buffering bodies.if (isAwaitingBody) {CurrentPassport passport = CurrentPassport.fromSessionContext(inMesg.getContext());if (inMesg.hasCompleteBody()) {if (inMesg instanceof HttpRequestMessage) {passport.addIfNotAlready(PassportState.FILTERS_INBOUND_BUF_END);} else if (inMesg instanceof HttpResponseMessage) {passport.addIfNotAlready(PassportState.FILTERS_OUTBOUND_BUF_END);}}else {if (inMesg instanceof HttpRequestMessage) {passport.addIfNotAlready(PassportState.FILTERS_INBOUND_BUF_START);} else if (inMesg instanceof HttpResponseMessage) {passport.addIfNotAlready(PassportState.FILTERS_OUTBOUND_BUF_START);}}}if (isAwaitingBody && inMesg.hasCompleteBody()) {//whole body has arrived, resume filter chain
                    runFilters(inMesg, runningFilterIdx);}}}catch (Exception ex) {handleException(inMesg, filterName, ex);}}

            if (filter.getSyncType() == FilterSyncType.SYNC) {final SyncZuulFilter<I, O> syncFilter = (SyncZuulFilter) filter;final O outMesg = syncFilter.apply(inMesg);recordFilterCompletion(SUCCESS, filter, startTime, inMesg, snapshot);return (outMesg != null) ? outMesg : filter.getDefaultOutput(inMesg);}// async filter
            filter.incrementConcurrency();resumer = new FilterChainResumer(inMesg, filter, snapshot, startTime);filter.applyAsync(inMesg).observeOn(Schedulers.from(getChannelHandlerContext(inMesg).executor())).doOnUnsubscribe(resumer::decrementConcurrency).subscribe(resumer);

    @Overridepublic HttpResponseMessage apply(final HttpRequestMessage input) {// If no Origin has been selected, then just return a 404 static response.// handle any exception heretry {if (origin == null) {handleNoOriginSelected();return null;}origin.getProxyTiming(zuulRequest).start();// To act the same as Ribbon, we must do this before starting execution (as well as before each attempt).IClientConfig requestConfig = origin.getExecutionContext(zuulRequest).getRequestConfig();originalReadTimeout = requestConfig.getProperty(ReadTimeout, null);setReadTimeoutOnContext(requestConfig, 1);origin.onRequestExecutionStart(zuulRequest); proxyRequestToOrigin();//Doesn't return origin response to caller, calls invokeNext() internally in response filter chainreturn null;} catch (Exception ex) {handleError(ex);return null;}}

    private void proxyRequestToOrigin() {Promise<PooledConnection> promise = null;try {attemptNum += 1;requestStat = createRequestStat();origin.preRequestChecks(zuulRequest);concurrentReqCount++;// update RPS trackers
            updateOriginRpsTrackers(origin, attemptNum);// We pass this AtomicReference<Server> here and the origin impl will assign the chosen server to it.promise = origin.connectToOrigin(zuulRequest, channelCtx.channel().eventLoop(), attemptNum, passport, chosenServer, chosenHostAddr);storeAndLogOriginRequestInfo();currentRequestAttempt = origin.newRequestAttempt(chosenServer.get(), context, attemptNum);requestAttempts.add(currentRequestAttempt);passport.add(PassportState.ORIGIN_CONN_ACQUIRE_START);if (promise.isDone()) {operationComplete(promise);} else {promise.addListener(this);}}catch (Exception ex) {LOG.error("Error while connecting to origin, UUID {} " + context.getUUID(), ex);storeAndLogOriginRequestInfo();if (promise != null && ! promise.isDone()) {promise.setFailure(ex);} else {errorFromOrigin(ex);}}}

    @Overridepublic Promise<PooledConnection> connectToOrigin(HttpRequestMessage zuulReq, EventLoop eventLoop, int attemptNumber,CurrentPassport passport, AtomicReference<Server> chosenServer,AtomicReference<String> chosenHostAddr) {return clientChannelManager.acquire(eventLoop, null, zuulReq.getMethod().toUpperCase(),zuulReq.getPath(), attemptNumber, passport, chosenServer, chosenHostAddr);}