import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
/**
 * Reset the deprecation logger by removing the current thread context, and setting a new thread context if {@code setNewThreadContext}
 * is set to {@code true} and otherwise clearing the current thread context.
 *
 * @param setNewThreadContext whether or not to attach a new thread context to the deprecation logger
 */
private void resetDeprecationLogger(final boolean setNewThreadContext) {
    // "clear" current warning headers by setting a new ThreadContext
    DeprecationLogger.removeThreadContext(this.threadContext);
    try {
        this.threadContext.close();
        // catch IOException to avoid that call sites have to deal with it. It is only declared because this class implements Closeable
        // but it is impossible that this implementation will ever throw an IOException.
    } catch (IOException ex) {
        throw new AssertionError("IOException thrown while closing deprecation logger's thread context", ex);
    }
    if (setNewThreadContext) {
        this.threadContext = new ThreadContext(Settings.EMPTY);
        DeprecationLogger.setThreadContext(this.threadContext);
    } else {
        this.threadContext = null;
    }
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
/**
 * Logs a deprecated message to the deprecation log, as well as to the local {@link ThreadContext}.
 *
 * @param threadContexts The node's {@link ThreadContext} (outside of concurrent tests, this should only ever have one context).
 * @param message The deprecation message.
 * @param params The parameters used to fill in the message, if any exist.
 */
@SuppressLoggerChecks(reason = "safely delegates to logger")
void deprecated(final Set<ThreadContext> threadContexts, final String message, final Object... params) {
    final Iterator<ThreadContext> iterator = threadContexts.iterator();

    if (iterator.hasNext()) {
        final String formattedMessage = LoggerMessageFormat.format(message, params);
        final String warningHeaderValue = formatWarning(formattedMessage);
        assert WARNING_HEADER_PATTERN.matcher(warningHeaderValue).matches();
        assert extractWarningValueFromWarningHeader(warningHeaderValue).equals(escape(formattedMessage));
        while (iterator.hasNext()) {
            try {
                final ThreadContext next = iterator.next();
                next.addResponseHeader("Warning", warningHeaderValue, DeprecationLogger::extractWarningValueFromWarningHeader);
            } catch (final IllegalStateException e) {
                // ignored; it should be removed shortly
            }
        }
        logger.warn(formattedMessage);
    } else {
        logger.warn(message, params);
    }
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
@Override
public List<TransportInterceptor> getTransportInterceptors(NamedWriteableRegistry namedWriteableRegistry,
                                                           ThreadContext threadContext) {
    return Collections.singletonList(new TransportInterceptor() {
        @Override
        public <T extends TransportRequest> TransportRequestHandler<T> interceptHandler(String action, String executor,
                                                                                    boolean forceExecution,
                                                                                    TransportRequestHandler<T> actualHandler) {
            return instance.interceptHandler(action, executor, forceExecution, actualHandler);
        }

        @Override
        public AsyncSender interceptSender(AsyncSender sender) {
            return instance.interceptSender(sender);
        }
    });
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
public void testApplyRelevantHeaders() throws Exception {
    final ThreadContext threadContext = new ThreadContext(Settings.EMPTY);
    Set<String> headers = new HashSet<>(Arrays.asList("header.1", "header.2"));
    final RestController restController = new RestController(Settings.EMPTY, headers, null, null, circuitBreakerService);
    Map<String, List<String>> restHeaders = new HashMap<>();
    restHeaders.put("header.1", Collections.singletonList("true"));
    restHeaders.put("header.2", Collections.singletonList("true"));
    restHeaders.put("header.3", Collections.singletonList("false"));
    restController.dispatchRequest(new FakeRestRequest.Builder(xContentRegistry()).withHeaders(restHeaders).build(), null, null,
        threadContext, (RestRequest request, RestChannel channel, NodeClient client) -> {
            assertEquals("true", threadContext.getHeader("header.1"));
            assertEquals("true", threadContext.getHeader("header.2"));
            assertNull(threadContext.getHeader("header.3"));
        });
    // the rest controller relies on the caller to stash the context, so we should expect these values here as we didn't stash the
    // context in this test
    assertEquals("true", threadContext.getHeader("header.1"));
    assertEquals("true", threadContext.getHeader("header.2"));
    assertNull(threadContext.getHeader("header.3"));
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
public void testAddsHeaderWithThreadContext() throws IOException {
    try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) {
        final Set<ThreadContext> threadContexts = Collections.singleton(threadContext);

        final String param = randomAsciiOfLengthBetween(1, 5);
        logger.deprecated(threadContexts, "A simple message [{}]", param);

        final Map<String, List<String>> responseHeaders = threadContext.getResponseHeaders();

        assertThat(responseHeaders.size(), equalTo(1));
        final List<String> responses = responseHeaders.get("Warning");
        assertThat(responses, hasSize(1));
        assertThat(responses.get(0), warningValueMatcher);
        assertThat(responses.get(0), containsString("\"A simple message [" + param + "]\""));
    }
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
public void testAddsCombinedHeaderWithThreadContext() throws IOException {
    try (ThreadContext threadContext = new ThreadContext(Settings.EMPTY)) {
        final Set<ThreadContext> threadContexts = Collections.singleton(threadContext);

        final String param = randomAsciiOfLengthBetween(1, 5);
        logger.deprecated(threadContexts, "A simple message [{}]", param);
        final String second = randomAsciiOfLengthBetween(1, 10);
        logger.deprecated(threadContexts, second);

        final Map<String, List<String>> responseHeaders = threadContext.getResponseHeaders();

        assertEquals(1, responseHeaders.size());

        final List<String> responses = responseHeaders.get("Warning");

        assertEquals(2, responses.size());
        assertThat(responses.get(0), warningValueMatcher);
        assertThat(responses.get(0), containsString("\"A simple message [" + param + "]\""));
        assertThat(responses.get(1), warningValueMatcher);
        assertThat(responses.get(1), containsString("\"" + second + "\""));
    }
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
public static String getSafeFromHeader(final ThreadContext context, final String headerName) {

        if (context == null || headerName == null || headerName.isEmpty()) {
            return null;
        }

        String headerValue = null;
        	
        Map<String, String> headers = context.getHeaders();
        if (!headers.containsKey(headerName) || (headerValue = headers.get(headerName)) == null) {
            return null;
        }

        if (isInterClusterRequest(context) || isDirectRequest(context)) {
            return headerValue;
        }

        return null;
    }
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
private SearchGuardLicense createOrGetTrial(String msg) {
    long created = System.currentTimeMillis();
    ThreadContext threadContext = threadPool.getThreadContext();
    
    try(StoredContext ctx = threadContext.stashContext()) {
        threadContext.putHeader(ConfigConstants.SG_CONF_REQUEST_HEADER, "true");
        GetResponse get = client.prepareGet(searchguardIndex, "sg", "tattr").get();
        if(get.isExists()) {
            created = (long) get.getSource().get("val");
        } else {
            client.index(new IndexRequest(searchguardIndex)
            .type("sg")
            .id("tattr")
            .setRefreshPolicy(RefreshPolicy.IMMEDIATE)
            .source("{\"val\": "+System.currentTimeMillis()+"}", XContentType.JSON)).actionGet();
        }
    }
    
    return SearchGuardLicense.createTrialLicense(formatDate(created), clusterService, msg);
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
public void testAuthenticateBadUser() {
    Settings globalSettings = Settings.builder().put("path.home", createTempDir()).build();
    Settings realmSettings = Settings.builder()
            .put("type", CustomRealm.TYPE)
            .put("users.john.password", "doe")
            .put("users.john.roles", "user")
            .put("users.jane.password", "test")
            .putList("users.jane.roles", "user", "admin")
            .build();

    CustomRealm realm = new CustomRealm(new RealmConfig("test", realmSettings, globalSettings,
            new Environment(globalSettings, createTempDir()), new ThreadContext(globalSettings)));
    UsernamePasswordToken token =
            new UsernamePasswordToken("john1", new SecureString(randomAlphaOfLengthBetween(4, 16).toCharArray()));
    realm.authenticate(token, ActionListener.wrap(result -> {
        assertFalse(result.isAuthenticated());
        assertThat(result.getUser(), nullValue());
    }, e -> fail("Failed with exception: " + e.getMessage())));
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
public void testTransportClient() throws Exception {
    NodesInfoResponse nodeInfos = client().admin().cluster().prepareNodesInfo().get();
    List<NodeInfo>  nodes = nodeInfos.getNodes();
    assertTrue(nodes.size() > 0);
    TransportAddress publishAddress = randomFrom(nodes).getTransport().address().publishAddress();
    String clusterName = nodeInfos.getClusterName().value();

    Settings settings = Settings.builder()
            .put("cluster.name", clusterName)
            .put(ThreadContext.PREFIX + "." + CustomRealm.USER_HEADER, randomFrom(KNOWN_USERS))
            .put(ThreadContext.PREFIX + "." + CustomRealm.PW_HEADER, PASSWORD)
            .build();
    try (TransportClient client = new PreBuiltXPackTransportClient(settings)) {
        client.addTransportAddress(publishAddress);
        ClusterHealthResponse response = client.admin().cluster().prepareHealth().execute().actionGet();
        assertThat(response.isTimedOut(), is(false));
    }
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
/**
 * @param transport             The corresponding <code>NettyHttpServerTransport</code> where this channel belongs to.
 * @param request               The request that is handled by this channel.
 * @param pipelinedRequest      If HTTP pipelining is enabled provide the corresponding pipelined request. May be null if
 *                              HTTP pipelining is disabled.
 * @param detailedErrorsEnabled true iff error messages should include stack traces.
 * @param threadContext         the thread context for the channel
 */
Netty4HttpChannel(
        final Netty4HttpServerTransport transport,
        final Netty4HttpRequest request,
        final HttpPipelinedRequest pipelinedRequest,
        final boolean detailedErrorsEnabled,
        final ThreadContext threadContext) {
    super(request, detailedErrorsEnabled);
    this.transport = transport;
    this.channel = request.getChannel();
    this.nettyRequest = request.request();
    this.pipelinedRequest = pipelinedRequest;
    this.threadContext = threadContext;
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
protected HttpChannelHandler(
        final Netty4HttpServerTransport transport,
        final boolean detailedErrorsEnabled,
        final ThreadContext threadContext) {
    this.transport = transport;
    this.requestHandler = new Netty4HttpRequestHandler(transport, detailedErrorsEnabled, threadContext);
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
@Before
public final void before()  {
    logger.info("[{}]: before test", getTestName());
    assertNull("Thread context initialized twice", threadContext);
    if (enableWarningsCheck()) {
        this.threadContext = new ThreadContext(Settings.EMPTY);
        DeprecationLogger.setThreadContext(threadContext);
    }
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
protected RestClient buildClient(Settings settings, HttpHost[] hosts) throws IOException {
    RestClientBuilder builder = RestClient.builder(hosts);
    String keystorePath = settings.get(TRUSTSTORE_PATH);
    if (keystorePath != null) {
        final String keystorePass = settings.get(TRUSTSTORE_PASSWORD);
        if (keystorePass == null) {
            throw new IllegalStateException(TRUSTSTORE_PATH + " is provided but not " + TRUSTSTORE_PASSWORD);
        }
        Path path = PathUtils.get(keystorePath);
        if (!Files.exists(path)) {
            throw new IllegalStateException(TRUSTSTORE_PATH + " is set but points to a non-existing file");
        }
        try {
            KeyStore keyStore = KeyStore.getInstance("jks");
            try (InputStream is = Files.newInputStream(path)) {
                keyStore.load(is, keystorePass.toCharArray());
            }
            SSLContext sslcontext = SSLContexts.custom().loadTrustMaterial(keyStore, null).build();
            SSLIOSessionStrategy sessionStrategy = new SSLIOSessionStrategy(sslcontext);
            builder.setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder.setSSLStrategy(sessionStrategy));
        } catch (KeyStoreException|NoSuchAlgorithmException|KeyManagementException|CertificateException e) {
            throw new RuntimeException("Error setting up ssl", e);
        }
    }

    try (ThreadContext threadContext = new ThreadContext(settings)) {
        Header[] defaultHeaders = new Header[threadContext.getHeaders().size()];
        int i = 0;
        for (Map.Entry<String, String> entry : threadContext.getHeaders().entrySet()) {
            defaultHeaders[i++] = new BasicHeader(entry.getKey(), entry.getValue());
        }
        builder.setDefaultHeaders(defaultHeaders);
    }
    return builder.build();
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
public Parameters(Environment env, ScriptService scriptService, TemplateService templateService,
                  AnalysisRegistry analysisRegistry, ThreadContext threadContext) {
    this.env = env;
    this.scriptService = scriptService;
    this.templateService = templateService;
    this.threadContext = threadContext;
    this.analysisRegistry = analysisRegistry;
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
@Override
public Client filterWithHeader(Map<String, String> headers) {
    return new FilterClient(this) {
        @Override
        protected <Request extends ActionRequest, Response extends ActionResponse, RequestBuilder extends ActionRequestBuilder<Request, Response, RequestBuilder>> void doExecute(Action<Request, Response, RequestBuilder> action, Request request, ActionListener<Response> listener) {
            ThreadContext threadContext = threadPool().getThreadContext();
            try (ThreadContext.StoredContext ctx = threadContext.stashAndMergeHeaders(headers)) {
                super.doExecute(action, request, listener);
            }
        }
    };
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
@Override
public void dispatchRequest(RestRequest request, RestChannel channel, ThreadContext threadContext) {
    if (request.rawPath().equals("/favicon.ico")) {
        handleFavicon(request, channel);
        return;
    }
    RestChannel responseChannel = channel;
    try {
        final int contentLength = request.hasContent() ? request.content().length() : 0;
        assert contentLength >= 0 : "content length was negative, how is that possible?";
        final RestHandler handler = getHandler(request);

        if (contentLength > 0 && hasContentType(request, handler) == false) {
            sendContentTypeErrorMessage(request, responseChannel);
        } else if (contentLength > 0 && handler != null && handler.supportsContentStream() &&
            request.getXContentType() != XContentType.JSON && request.getXContentType() != XContentType.SMILE) {
            responseChannel.sendResponse(BytesRestResponse.createSimpleErrorResponse(RestStatus.NOT_ACCEPTABLE, "Content-Type [" +
                request.getXContentType() + "] does not support stream parsing. Use JSON or SMILE instead"));
        } else {
            if (canTripCircuitBreaker(request)) {
                inFlightRequestsBreaker(circuitBreakerService).addEstimateBytesAndMaybeBreak(contentLength, "<http_request>");
            } else {
                inFlightRequestsBreaker(circuitBreakerService).addWithoutBreaking(contentLength);
            }
            // iff we could reserve bytes for the request we need to send the response also over this channel
            responseChannel = new ResourceHandlingHttpChannel(channel, circuitBreakerService, contentLength);
            dispatchRequest(request, responseChannel, client, threadContext, handler);
        }
    } catch (Exception e) {
        try {
            responseChannel.sendResponse(new BytesRestResponse(channel, e));
        } catch (Exception inner) {
            inner.addSuppressed(e);
            logger.error((Supplier<?>) () ->
                new ParameterizedMessage("failed to send failure response for uri [{}]", request.uri()), inner);
        }
    }
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
/**
 * @param timeout        a global timeout for this observer. After it has expired the observer
 *                       will fail any existing or new #waitForNextChange calls. Set to null
 *                       to wait indefinitely
 */
public ClusterStateObserver(ClusterState initialState, ClusterService clusterService, @Nullable TimeValue timeout, Logger logger,
                            ThreadContext contextHolder) {
    this.clusterService = clusterService;
    this.lastObservedState = new AtomicReference<>(new StoredState(initialState));
    this.timeOutValue = timeout;
    if (timeOutValue != null) {
        this.startTimeNS = System.nanoTime();
    }
    this.logger = logger;
    this.contextHolder = contextHolder;
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
/**
 * Set the {@link ThreadContext} used to add deprecation headers to network responses.
 * <p>
 * This is expected to <em>only</em> be invoked by the {@code Node}'s constructor (therefore once outside of tests).
 *
 * @param threadContext The thread context owned by the {@code ThreadPool} (and implicitly a {@code Node})
 * @throws IllegalStateException if this {@code threadContext} has already been set
 */
public static void setThreadContext(ThreadContext threadContext) {
    Objects.requireNonNull(threadContext, "Cannot register a null ThreadContext");

    // add returning false means it _did_ have it already
    if (THREAD_CONTEXT.add(threadContext) == false) {
        throw new IllegalStateException("Double-setting ThreadContext not allowed!");
    }
}
 

import org.elasticsearch.common.util.concurrent.ThreadContext; //导入依赖的package包/类
/**
 * Remove the {@link ThreadContext} used to add deprecation headers to network responses.
 * <p>
 * This is expected to <em>only</em> be invoked by the {@code Node}'s {@code close} method (therefore once outside of tests).
 *
 * @param threadContext The thread context owned by the {@code ThreadPool} (and implicitly a {@code Node})
 * @throws IllegalStateException if this {@code threadContext} is unknown (and presumably already unset before)
 */
public static void removeThreadContext(ThreadContext threadContext) {
    assert threadContext != null;

    // remove returning false means it did not have it already
    if (THREAD_CONTEXT.remove(threadContext) == false) {
        throw new IllegalStateException("Removing unknown ThreadContext not allowed!");
    }
}