Package com.linkedin.alpini.router.api

Class ScatterGatherHelper<H,P extends ResourcePath<K>,K,R,BASIC_HTTP_REQUEST extends BasicRequest,HTTP_RESPONSE,HTTP_RESPONSE_STATUS>

java.lang.Object

com.linkedin.alpini.router.api.ScatterGatherHelper<H,P,K,R,BASIC_HTTP_REQUEST,HTTP_RESPONSE,HTTP_RESPONSE_STATUS>

public class ScatterGatherHelper<H,P extends ResourcePath<K>,K,R,BASIC_HTTP_REQUEST extends BasicRequest,HTTP_RESPONSE,HTTP_RESPONSE_STATUS>
extends Object

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	ScatterGatherHelper.Builder<H,P extends ResourcePath<K>,K,R,HTTP_REQUEST extends BasicRequest,HTTP_RESPONSE,HTTP_RESPONSE_STATUS>

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	ScatterGatherHelper(ExtendedResourcePathParser<P,K,BASIC_HTTP_REQUEST> pathParser, AsyncPartitionFinder<K> partitionFinder, HostFinder<H,R> hostFinder, HostHealthMonitor<H> hostHealthMonitor, RoleFinder<R> roleFinder, ScatterGatherMode broadcastMode, ScatterGatherMode scatterMode, PartitionDispatchHandler<H,P,K,BASIC_HTTP_REQUEST,HTTP_RESPONSE,HTTP_RESPONSE_STATUS> dispatchHandler, Optional<ResponseAggregatorFactory<BASIC_HTTP_REQUEST,HTTP_RESPONSE>> responseAggregatorFactory, Function<Headers,Long> requestTimeout, LongTailRetrySupplier<P,K> longTailRetrySupplier, Function<BasicRequest,Metrics> metricsProvider, BiFunction<Headers,Metrics,Headers> metricsDecorator, Function<Headers,Metrics> responseMetrics, Function<P,ScatterGatherStats> scatterGatherStatsProvider, long defaultTimeoutMillis, long dispatchMinimumMillis, long longTailMinimumMillis, List<Pair<String,Supplier<?>>> beforeHttpServerCodec, List<Pair<String,Supplier<?>>> beforeChunkAggregator, List<Pair<String,Supplier<?>>> beforeIdleStateHandler, List<Pair<String,Supplier<?>>> beforeHttpRequestHandler, IntPredicate successCodePredicate, RequestRetriableChecker<P,R,HTTP_RESPONSE_STATUS> requestRetriableChecker, boolean enableStackTraceResponseForException, BooleanSupplier enableRetryRequestAlwaysUseADifferentHost, BooleanSupplier disableRetryOnTimeout, BooleanSupplier isReqRedirectionAllowedForQuery)

Method Summary

Modifier and Type	Method	Description
HTTP_RESPONSE	aggregateResponse(BASIC_HTTP_REQUEST request, Metrics metrics, List<HTTP_RESPONSE> responses, ResponseAggregatorFactory<BASIC_HTTP_REQUEST,HTTP_RESPONSE> defaultAggregator)
static ScatterGatherHelper.Builder<?,?,?,?,?,?,?>	builder()
void	decorateResponse(Headers responseHeaders, Headers requestHeaders, Metrics metrics)
boolean	disableRetryOnTimeout()
final void	dispatch(Scatter<H,P,K> scatter, ScatterGatherRequest<H,K> part, P path, BASIC_HTTP_REQUEST request, AsyncPromise<H> hostSelected, AsyncPromise<List<HTTP_RESPONSE>> responseFuture, AsyncPromise<HTTP_RESPONSE_STATUS> retryFuture, AsyncFuture<Void> timeoutFuture, Executor contextExecutor)
Netty	dispatcherNettyVersion()
CompletionStage<String>	findPartitionName(String resourceName, K key)
void	forEachBeforeChunkAggregator(Consumer<Pair<String,?>> consumer)
void	forEachBeforeHttpRequestHandler(Consumer<Pair<String,?>> consumer)
void	forEachBeforeHttpServerCodec(Consumer<Pair<String,?>> consumer)
void	forEachBeforeIdleStateHandler(Consumer<Pair<String,?>> consumer)
long	getDefaultTimeoutMillis()
long	getDispatchMinimumMillis()
long	getLongTailMinimumMillis()
final AsyncFuture<LongSupplier>	getLongTailRetryMilliseconds(P path, BasicRequest request)
final AsyncFuture<LongSupplier>	getLongTailRetryMilliseconds(P path, String methodName)
long	getRequestTimeout(Headers headers)
ScatterGatherStats	getScatterGatherStatsByPath(P path)
Metrics	initializeMetrics(BasicRequest request)
boolean	isEnableRetryRequestAlwaysUseADifferentHost()
boolean	isEnableStackTraceResponseForException()
final boolean	isHostHealthy(H hostName, String partitionName)	To check if a host is healthy/responsive.
boolean	isReqRedirectionAllowedForQuery()
boolean	isRequestRetriable(P path, R role, HTTP_RESPONSE_STATUS status)
boolean	isSuccessStatus(int code)
final P	parseResourceUri(String uri, BASIC_HTTP_REQUEST request)
R	parseRoles(String requestMethod, Headers headers)
Metrics	responseMetrics(Headers headers)
CompletionStage<Scatter<H,P,K>>	scatter(String requestMethod, P path, Headers headers, HostHealthMonitor<H> hostHealthMonitor, Metrics metrics, String initialHost)
void	setDefaultTimeoutMillis(long timeoutMillis)
void	setDispatchMinimumMillis(long value)
void	setLongTailMinimumMillis(long value)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

ScatterGatherHelper

protected ScatterGatherHelper(@Nonnull
 ExtendedResourcePathParser<P,K,BASIC_HTTP_REQUEST> pathParser,
 @Nonnull
 AsyncPartitionFinder<K> partitionFinder,
 @Nonnull
 HostFinder<H,R> hostFinder,
 @Nonnull
 HostHealthMonitor<H> hostHealthMonitor,
 @Nonnull
 RoleFinder<R> roleFinder,
 @Nonnull
 ScatterGatherMode broadcastMode,
 @Nonnull
 ScatterGatherMode scatterMode,
 @Nonnull
 PartitionDispatchHandler<H,P,K,BASIC_HTTP_REQUEST,HTTP_RESPONSE,HTTP_RESPONSE_STATUS> dispatchHandler,
 @Nonnull
 Optional<ResponseAggregatorFactory<BASIC_HTTP_REQUEST,HTTP_RESPONSE>> responseAggregatorFactory,
 @Nonnull
 Function<Headers,Long> requestTimeout,
 @Nonnull
 LongTailRetrySupplier<P,K> longTailRetrySupplier,
 @Nonnull
 Function<BasicRequest,Metrics> metricsProvider,
 @Nonnull
 BiFunction<Headers,Metrics,Headers> metricsDecorator,
 @Nonnull
 Function<Headers,Metrics> responseMetrics,
 @Nonnull
 Function<P,ScatterGatherStats> scatterGatherStatsProvider,
 long defaultTimeoutMillis,
 long dispatchMinimumMillis,
 long longTailMinimumMillis,
 @Nonnull
 List<Pair<String,Supplier<?>>> beforeHttpServerCodec,
 @Nonnull
 List<Pair<String,Supplier<?>>> beforeChunkAggregator,
 @Nonnull
 List<Pair<String,Supplier<?>>> beforeIdleStateHandler,
 @Nonnull
 List<Pair<String,Supplier<?>>> beforeHttpRequestHandler,
 @Nonnull
 IntPredicate successCodePredicate,
 @Nonnull
 RequestRetriableChecker<P,R,HTTP_RESPONSE_STATUS> requestRetriableChecker,
 boolean enableStackTraceResponseForException,
 @Nonnull
 BooleanSupplier enableRetryRequestAlwaysUseADifferentHost,
 @Nonnull
 BooleanSupplier disableRetryOnTimeout,
 @Nonnull
 BooleanSupplier isReqRedirectionAllowedForQuery)

Method Details

forEachBeforeHttpServerCodec

public void forEachBeforeHttpServerCodec(@Nonnull
 Consumer<Pair<String,?>> consumer)

forEachBeforeChunkAggregator

public void forEachBeforeChunkAggregator(@Nonnull
 Consumer<Pair<String,?>> consumer)

forEachBeforeIdleStateHandler

public void forEachBeforeIdleStateHandler(@Nonnull
 Consumer<Pair<String,?>> consumer)

forEachBeforeHttpRequestHandler

public void forEachBeforeHttpRequestHandler(@Nonnull
 Consumer<Pair<String,?>> consumer)

parseResourceUri

@Nonnull
public final P parseResourceUri(@Nonnull
 String uri,
 @Nonnull
 BASIC_HTTP_REQUEST request)
                         throws RouterException

Throws:

RouterException

isHostHealthy

public final boolean isHostHealthy(@Nonnull
 H hostName,
 @Nonnull
 String partitionName)

To check if a host is healthy/responsive.

Parameters:

hostName - the host name, including the service port

Returns:

the host is healthy or not

getLongTailRetryMilliseconds

public final AsyncFuture<LongSupplier> getLongTailRetryMilliseconds(P path,
 String methodName)

getLongTailRetryMilliseconds

public final AsyncFuture<LongSupplier> getLongTailRetryMilliseconds(P path,
 BasicRequest request)

dispatch

public final void dispatch(@Nonnull
 Scatter<H,P,K> scatter,
 @Nonnull
 ScatterGatherRequest<H,K> part,
 @Nonnull
 P path,
 @Nonnull
 BASIC_HTTP_REQUEST request,
 @Nonnull
 AsyncPromise<H> hostSelected,
 @Nonnull
 AsyncPromise<List<HTTP_RESPONSE>> responseFuture,
 @Nonnull
 AsyncPromise<HTTP_RESPONSE_STATUS> retryFuture,
 @Nonnull
 AsyncFuture<Void> timeoutFuture,
 @Nonnull
 Executor contextExecutor)
                    throws RouterException

Throws:

RouterException

dispatcherNettyVersion

public Netty dispatcherNettyVersion()

scatter

@Nonnull
public CompletionStage<Scatter<H,P,K>> scatter(@Nonnull
 String requestMethod,
 @Nonnull
 P path,
 @Nonnull
 Headers headers,
 @Nonnull
 HostHealthMonitor<H> hostHealthMonitor,
 Metrics metrics,
 String initialHost)
                                        throws RouterException

Throws:

RouterException

aggregateResponse

@Nonnull
public HTTP_RESPONSE aggregateResponse(@Nonnull
 BASIC_HTTP_REQUEST request,
 Metrics metrics,
 @Nonnull
 List<HTTP_RESPONSE> responses,
 @Nonnull
 ResponseAggregatorFactory<BASIC_HTTP_REQUEST,HTTP_RESPONSE> defaultAggregator)

findPartitionName

public CompletionStage<String> findPartitionName(String resourceName,
 K key)

getRequestTimeout

public long getRequestTimeout(@Nonnull
 Headers headers)

getDefaultTimeoutMillis

public long getDefaultTimeoutMillis()

setDefaultTimeoutMillis

public void setDefaultTimeoutMillis(long timeoutMillis)

getDispatchMinimumMillis

public long getDispatchMinimumMillis()

setDispatchMinimumMillis

public void setDispatchMinimumMillis(long value)

getLongTailMinimumMillis

public long getLongTailMinimumMillis()

setLongTailMinimumMillis

public void setLongTailMinimumMillis(long value)

isEnableStackTraceResponseForException

public boolean isEnableStackTraceResponseForException()

isEnableRetryRequestAlwaysUseADifferentHost

public boolean isEnableRetryRequestAlwaysUseADifferentHost()

disableRetryOnTimeout

public boolean disableRetryOnTimeout()

isReqRedirectionAllowedForQuery

public boolean isReqRedirectionAllowedForQuery()

decorateResponse

public void decorateResponse(@Nonnull
 Headers responseHeaders,
 @Nonnull
 Headers requestHeaders,
 Metrics metrics)

initializeMetrics

public Metrics initializeMetrics(@Nonnull
 BasicRequest request)

responseMetrics

public Metrics responseMetrics(@Nonnull
 Headers headers)

getScatterGatherStatsByPath

@Nonnull
public ScatterGatherStats getScatterGatherStatsByPath(P path)

builder

public static ScatterGatherHelper.Builder<?,?,?,?,?,?,?> builder()

isSuccessStatus

public boolean isSuccessStatus(int code)

isRequestRetriable

public boolean isRequestRetriable(@Nonnull
 P path,
 @Nonnull
 R role,
 @Nonnull
 HTTP_RESPONSE_STATUS status)

parseRoles

public R parseRoles(@Nonnull
 String requestMethod,
 @Nonnull
 Headers headers)