text:
package org.elasticsearch.action.support.replication;

import org.apache.logging.log4j.message.ParameterizedMessage;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.ActionListenerResponseHandler;
import org.elasticsearch.action.ActionResponse;
import org.elasticsearch.action.UnavailableShardsException;
import org.elasticsearch.action.support.ActionFilters;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.action.support.TransportAction;
import org.elasticsearch.action.support.TransportActions;
import org.elasticsearch.client.transport.NoNodeAvailableException;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateObserver;
import org.elasticsearch.cluster.action.shard.ShardStateAction;
import org.elasticsearch.cluster.block.ClusterBlockException;
import org.elasticsearch.cluster.block.ClusterBlockLevel;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.cluster.routing.AllocationId;
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.lease.Releasable;
import org.elasticsearch.common.lease.Releasables;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.util.concurrent.AbstractRunnable;
import org.elasticsearch.common.util.concurrent.ThreadContext;
import org.elasticsearch.index.IndexNotFoundException;
import org.elasticsearch.index.IndexService;
import org.elasticsearch.index.shard.IndexShard;
import org.elasticsearch.index.shard.IndexShardState;
import org.elasticsearch.index.shard.ShardId;
import org.elasticsearch.index.shard.ShardNotFoundException;
import org.elasticsearch.indices.IndicesService;
import org.elasticsearch.node.NodeClosedException;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.tasks.TaskId;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.ConnectTransportException;
import org.elasticsearch.transport.TransportChannel;
import org.elasticsearch.transport.TransportChannelResponseHandler;
import org.elasticsearch.transport.TransportException;
import org.elasticsearch.transport.TransportRequest;
import org.elasticsearch.transport.TransportRequestHandler;
import org.elasticsearch.transport.TransportRequestOptions;
import org.elasticsearch.transport.TransportResponse;
import org.elasticsearch.transport.TransportResponse.Empty;
import org.elasticsearch.transport.TransportResponseHandler;
import org.elasticsearch.transport.TransportService;

import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * Base class for requests that should be executed on a primary copy followed by replica copies.
 * Subclasses can resolve the target shard and provide implementation for primary and replica operations.
 *
 * The action samples cluster state on the receiving node to reroute to node with primary copy and on the
 * primary node to validate request before primary operation followed by sampling state again for resolving
 * nodes with replica copies to perform replication.
 */
public abstract class TransportReplicationAction<
            Request extends ReplicationRequest<Request>,
            ReplicaRequest extends ReplicationRequest<ReplicaRequest>,
            Response extends ReplicationResponse
        > extends TransportAction<Request, Response> {

    private final TransportService transportService;
    protected final ClusterService clusterService;
    private final IndicesService indicesService;
    private final ShardStateAction shardStateAction;
    private final TransportRequestOptions transportOptions;
    private final String executor;

    // package private for testing
    private final String transportReplicaAction;
    private final String transportPrimaryAction;
    private final ReplicasProxy replicasProxy;

    protected TransportReplicationAction(Settings settings, String actionName, TransportService transportService,
                                         ClusterService clusterService, IndicesService indicesService,
                                         ThreadPool threadPool, ShardStateAction shardStateAction,
                                         ActionFilters actionFilters,
                                         IndexNameExpressionResolver indexNameExpressionResolver, Supplier<Request> request,
                                         Supplier<ReplicaRequest> replicaRequest, String executor) {
        super(settings, actionName, threadPool, actionFilters, indexNameExpressionResolver, transportService.getTaskManager());
        this.transportService = transportService;
        this.clusterService = clusterService;
        this.indicesService = indicesService;
        this.shardStateAction = shardStateAction;
        this.executor = executor;

        this.transportPrimaryAction = actionName + "[p]";
        this.transportReplicaAction = actionName + "[r]";
        transportService.registerRequestHandler(actionName, request, ThreadPool.Names.SAME, new OperationTransportHandler());
        transportService.registerRequestHandler(transportPrimaryAction, () -> new ConcreteShardRequest<>(request), executor,
            new PrimaryOperationTransportHandler());
        // we must never reject on because of thread pool capacity on replicas
        transportService.registerRequestHandler(transportReplicaAction,
            () -> new ConcreteShardRequest<>(replicaRequest),
            executor, true, true,
            new ReplicaOperationTransportHandler());

        this.transportOptions = transportOptions();

        this.replicasProxy = new ReplicasProxy();
    }

    @Override
    protected final void doExecute(Request request, ActionListener<Response> listener) {
        throw new UnsupportedOperationException("the task parameter is required for this operation");
    }

    @Override
    protected void doExecute(Task task, Request request, ActionListener<Response> listener) {
        new ReroutePhase((ReplicationTask) task, request, listener).run();
    }

    protected abstract Response newResponseInstance();

    /**
     * Resolves derived values in the request. For example, the target shard id of the incoming request,
     * if not set at request construction
     * Additional processing or validation of the request should be done here.
     *
     * @param metaData      cluster state metadata
     * @param indexMetaData index metadata of the concrete index this request is going to operate on
     * @param request       the request to resolve
     */
    protected void resolveRequest(MetaData metaData, IndexMetaData indexMetaData, Request request) {
        if (request.waitForActiveShards() == ActiveShardCount.DEFAULT) {
            // if the wait for active shard count has not been set in the request,
            // resolve it from the index settings
            request.waitForActiveShards(indexMetaData.getWaitForActiveShards());
        }
    }

    /**
     * Primary operation on node with primary copy.
     *
     * @param shardRequest the request to the primary shard
     * @param primary      the primary shard to perform the operation on
     */
    protected abstract PrimaryResult shardOperationOnPrimary(Request shardRequest, IndexShard primary) throws Exception;

    /**
     * Synchronous replica operation on nodes with replica copies. This is done under the lock form
     * {@link IndexShard#acquireReplicaOperationLock(long, ActionListener, String)}
     *
     * @param shardRequest the request to the replica shard
     * @param replica      the replica shard to perform the operation on
     */
    protected abstract ReplicaResult shardOperationOnReplica(ReplicaRequest shardRequest, IndexShard replica) throws Exception;

    /**
     * Cluster level block to check before request execution
     */
    protected ClusterBlockLevel globalBlockLevel() {
        return ClusterBlockLevel.WRITE;
    }

    /**
     * Index level block to check before request execution
     */
    protected ClusterBlockLevel indexBlockLevel() {
        return ClusterBlockLevel.WRITE;
    }

    /**
     * True if provided index should be resolved when resolving request
     */
    protected boolean resolveIndex() {
        return true;
    }

    protected TransportRequestOptions transportOptions() {
        return TransportRequestOptions.EMPTY;
    }

    protected boolean retryPrimaryException(final Throwable e) {
        return e.getClass() == ReplicationOperation.RetryOnPrimaryException.class
                || TransportActions.isShardNotAvailableException(e);
    }

    class OperationTransportHandler implements TransportRequestHandler<Request> {
        @Override
        public void messageReceived(final Request request, final TransportChannel channel, Task task) throws Exception {
            execute(task, request, new ActionListener<Response>() {
                @Override
                public void onResponse(Response result) {
                    try {
                        channel.sendResponse(result);
                    } catch (Exception e) {
                        onFailure(e);
                    }
                }

                @Override
                public void onFailure(Exception e) {
                    try {
                        channel.sendResponse(e);
                    } catch (Exception inner) {
                        inner.addSuppressed(e);
                        logger.warn(
                            (org.apache.logging.log4j.util.Supplier<?>)
                                () -> new ParameterizedMessage("Failed to send response for {}", actionName), inner);
                    }
                }
            });
        }

        @Override
        public void messageReceived(Request request, TransportChannel channel) throws Exception {
            throw new UnsupportedOperationException("the task parameter is required for this operation");
        }
    }

    class PrimaryOperationTransportHandler implements TransportRequestHandler<ConcreteShardRequest<Request>> {
        @Override
        public void messageReceived(final ConcreteShardRequest<Request> request, final TransportChannel channel) throws Exception {
            throw new UnsupportedOperationException("the task parameter is required for this operation");
        }

        @Override
        public void messageReceived(ConcreteShardRequest<Request> request, TransportChannel channel, Task task) {
            new AsyncPrimaryAction(request.request, request.targetAllocationID, channel, (ReplicationTask) task).run();
        }
    }

    class AsyncPrimaryAction extends AbstractRunnable implements ActionListener<PrimaryShardReference> {

        private final Request request;
        /** targetAllocationID of the shard this request is meant for */
        private final String targetAllocationID;
        private final TransportChannel channel;
        private final ReplicationTask replicationTask;

        AsyncPrimaryAction(Request request, String targetAllocationID, TransportChannel channel, ReplicationTask replicationTask) {
            this.request = request;
            this.targetAllocationID = targetAllocationID;
            this.channel = channel;
            this.replicationTask = replicationTask;
        }

        @Override
        protected void doRun() throws Exception {
            acquirePrimaryShardReference(request.shardId(), targetAllocationID, this);
        }

        @Override
        public void onResponse(PrimaryShardReference primaryShardReference) {
            try {
                if (primaryShardReference.isRelocated()) {
                    primaryShardReference.close(); // release shard operation lock as soon as possible
                    setPhase(replicationTask, "primary_delegation");
                    // delegate primary phase to relocation target
                    // it is safe to execute primary phase on relocation target as there are no more in-flight operations where primary
                    // phase is executed on local shard and all subsequent operations are executed on relocation target as primary phase.
                    final ShardRouting primary = primaryShardReference.routingEntry();
                    assert primary.relocating() : "indexShard is marked as relocated but routing isn't" + primary;
                    DiscoveryNode relocatingNode = clusterService.state().nodes().get(primary.relocatingNodeId());
                    transportService.sendRequest(relocatingNode, transportPrimaryAction,
                        new ConcreteShardRequest<>(request, primary.allocationId().getRelocationId()),
                        transportOptions,
                        new TransportChannelResponseHandler<Response>(logger, channel, "rerouting indexing to target primary " + primary,
                            TransportReplicationAction.this::newResponseInstance) {

                            @Override
                            public void handleResponse(Response response) {
                                setPhase(replicationTask, "finished");
                                super.handleResponse(response);
                            }

                            @Override
                            public void handleException(TransportException exp) {
                                setPhase(replicationTask, "finished");
                                super.handleException(exp);
                            }
                        });
                } else {
                    setPhase(replicationTask, "primary");
                    final IndexMetaData indexMetaData = clusterService.state().getMetaData().index(request.shardId().getIndex());
                    final boolean executeOnReplicas = (indexMetaData == null) || shouldExecuteReplication(indexMetaData.getSettings());
                    final ActionListener<Response> listener = createResponseListener(primaryShardReference);
                    createReplicatedOperation(request,
                            ActionListener.wrap(result -> result.respond(listener), listener::onFailure),
                            primaryShardReference, executeOnReplicas)
                            .execute();
                }
            } catch (Exception e) {
                Releasables.closeWhileHandlingException(primaryShardReference); // release shard operation lock before responding to caller
                onFailure(e);
            }
        }

        @Override
        public void onFailure(Exception e) {
            setPhase(replicationTask, "finished");
            try {
                channel.sendResponse(e);
            } catch (IOException inner) {
                inner.addSuppressed(e);
                logger.warn("failed to send response", inner);
            }
        }

        private ActionListener<Response> createResponseListener(final PrimaryShardReference primaryShardReference) {
            return new ActionListener<Response>() {
                @Override
                public void onResponse(Response response) {
                    primaryShardReference.close(); // release shard operation lock before responding to caller
                    setPhase(replicationTask, "finished");
                    try {
                        channel.sendResponse(response);
                    } catch (IOException e) {
                        onFailure(e);
                    }
                }

                @Override
                public void onFailure(Exception e) {
                    primaryShardReference.close(); // release shard operation lock before responding to caller
                    setPhase(replicationTask, "finished");
                    try {
                        channel.sendResponse(e);
                    } catch (IOException e1) {
                        logger.warn("failed to send response", e);
                    }
                }
            };
        }

        protected ReplicationOperation<Request, ReplicaRequest, PrimaryResult> createReplicatedOperation(
            Request request, ActionListener<PrimaryResult> listener,
            PrimaryShardReference primaryShardReference, boolean executeOnReplicas) {
            return new ReplicationOperation<>(request, primaryShardReference, listener,
                executeOnReplicas, replicasProxy, clusterService::state, logger, actionName
            );
        }
    }

    protected class PrimaryResult implements ReplicationOperation.PrimaryResult<ReplicaRequest> {
        final ReplicaRequest replicaRequest;
        final Response finalResponseIfSuccessful;
        final Exception finalFailure;

        /**
         * Result of executing a primary operation
         * expects finalResponseIfSuccessful or finalFailure to be not-null
         */
        public PrimaryResult(ReplicaRequest replicaRequest, Response finalResponseIfSuccessful, Exception finalFailure) {
            assert finalFailure != null ^ finalResponseIfSuccessful != null
                    : "either a response or a failure has to be not null, " +
                    "found [" + finalFailure + "] failure and ["+ finalResponseIfSuccessful + "] response";
            this.replicaRequest = replicaRequest;
            this.finalResponseIfSuccessful = finalResponseIfSuccessful;
            this.finalFailure = finalFailure;
        }

        public PrimaryResult(ReplicaRequest replicaRequest, Response replicationResponse) {
            this(replicaRequest, replicationResponse, null);
        }

        @Override
        public ReplicaRequest replicaRequest() {
            return replicaRequest;
        }

        @Override
        public void setShardInfo(ReplicationResponse.ShardInfo shardInfo) {
            if (finalResponseIfSuccessful != null) {
                finalResponseIfSuccessful.setShardInfo(shardInfo);
            }
        }

        public void respond(ActionListener<Response> listener) {
            if (finalResponseIfSuccessful != null) {
                listener.onResponse(finalResponseIfSuccessful);
            } else {
                listener.onFailure(finalFailure);
            }
        }
    }

    protected class ReplicaResult {
        final Exception finalFailure;

        public ReplicaResult(Exception finalFailure) {
            this.finalFailure = finalFailure;
        }

        public ReplicaResult() {
            this(null);
        }

        public void respond(ActionListener<TransportResponse.Empty> listener) {
            if (finalFailure == null) {
                listener.onResponse(TransportResponse.Empty.INSTANCE);
            } else {
                listener.onFailure(finalFailure);
            }
        }
    }

    class ReplicaOperationTransportHandler implements TransportRequestHandler<ConcreteShardRequest<ReplicaRequest>> {
        @Override
        public void messageReceived(final ConcreteShardRequest<ReplicaRequest> request, final TransportChannel channel)
            throws Exception {
            throw new UnsupportedOperationException("the task parameter is required for this operation");
        }

        @Override
        public void messageReceived(ConcreteShardRequest<ReplicaRequest> requestWithAID, TransportChannel channel, Task task)
            throws Exception {
            new AsyncReplicaAction(requestWithAID.request, requestWithAID.targetAllocationID, channel, (ReplicationTask) task).run();
        }
    }

    public static class RetryOnReplicaException extends ElasticsearchException {

        public RetryOnReplicaException(ShardId shardId, String msg) {
            super(msg);
            setShard(shardId);
        }

        public RetryOnReplicaException(StreamInput in) throws IOException {
            super(in);
        }
    }

    private final class AsyncReplicaAction extends AbstractRunnable implements ActionListener<Releasable> {
        private final ReplicaRequest request;
        // allocation id of the replica this request is meant for
        private final String targetAllocationID;
        private final TransportChannel channel;
        private final IndexShard replica;
        /**
         * The task on the node with the replica shard.
         */
        private final ReplicationTask task;
        // important: we pass null as a timeout as failing a replica is
        // something we want to avoid at all costs
        private final ClusterStateObserver observer = new ClusterStateObserver(clusterService, null, logger, threadPool.getThreadContext());

        AsyncReplicaAction(ReplicaRequest request, String targetAllocationID, TransportChannel channel, ReplicationTask task) {
            this.request = request;
            this.channel = channel;
            this.task = task;
            this.targetAllocationID = targetAllocationID;
            final ShardId shardId = request.shardId();
            assert shardId != null : "request shardId must be set";
            this.replica = getIndexShard(shardId);
        }

        @Override
        public void onResponse(Releasable releasable) {
            try {
                ReplicaResult replicaResult = shardOperationOnReplica(request, replica);
                releasable.close(); // release shard operation lock before responding to caller
                final TransportReplicationAction.ReplicaResponse response =
                    new ReplicaResponse(replica.routingEntry().allocationId().getId(), replica.getLocalCheckpoint());
                replicaResult.respond(new ResponseListener(response));
            } catch (Exception e) {
                Releasables.closeWhileHandlingException(releasable); // release shard operation lock before responding to caller
                AsyncReplicaAction.this.onFailure(e);
            }
        }

        @Override
        public void onFailure(Exception e) {
            if (e instanceof RetryOnReplicaException) {
                logger.trace(
                    (org.apache.logging.log4j.util.Supplier<?>)
                        () -> new ParameterizedMessage(
                            "Retrying operation on replica, action [{}], request [{}]",
                            transportReplicaAction,
                            request),
                    e);
                request.onRetry();
                final ThreadContext.StoredContext context = threadPool.getThreadContext().newStoredContext();
                observer.waitForNextChange(new ClusterStateObserver.Listener() {
                    @Override
                    public void onNewClusterState(ClusterState state) {
                        context.close();
                        // Forking a thread on local node via transport service so that custom transport service have an
                        // opportunity to execute custom logic before the replica operation begins
                        String extraMessage = "action [" + transportReplicaAction + "], request[" + request + "]";
                        TransportChannelResponseHandler<TransportResponse.Empty> handler =
                            new TransportChannelResponseHandler<>(logger, channel, extraMessage, () -> TransportResponse.Empty.INSTANCE);
                        transportService.sendRequest(clusterService.localNode(), transportReplicaAction,
                            new ConcreteShardRequest<>(request, targetAllocationID),
                            handler);
                    }

                    @Override
                    public void onClusterServiceClose() {
                        responseWithFailure(new NodeClosedException(clusterService.localNode()));
                    }

                    @Override
                    public void onTimeout(TimeValue timeout) {
                        throw new AssertionError("Cannot happen: there is not timeout");
                    }
                });
            } else {
                responseWithFailure(e);
            }
        }

        protected void responseWithFailure(Exception e) {
            try {
                setPhase(task, "finished");
                channel.sendResponse(e);
            } catch (IOException responseException) {
                responseException.addSuppressed(e);
                logger.warn(
                    (org.apache.logging.log4j.util.Supplier<?>)
                        () -> new ParameterizedMessage(
                            "failed to send error message back to client for action [{}]",
                            transportReplicaAction),
                    responseException);
            }
        }

        @Override
        protected void doRun() throws Exception {
            setPhase(task, "replica");
            final String actualAllocationId = this.replica.routingEntry().allocationId().getId();
            if (actualAllocationId.equals(targetAllocationID) == false) {
                throw new ShardNotFoundException(this.replica.shardId(), "expected aID [{}] but found [{}]", targetAllocationID,
                    actualAllocationId);
            }
            replica.acquireReplicaOperationLock(request.primaryTerm, this, executor);
        }

        /**
         * Listens for the response on the replica and sends the response back to the primary.
         */
        private class ResponseListener implements ActionListener<TransportResponse.Empty> {
            private final ReplicaResponse replicaResponse;

            public ResponseListener(ReplicaResponse replicaResponse) {
                this.replicaResponse = replicaResponse;
            }

            @Override
            public void onResponse(Empty response) {
                if (logger.isTraceEnabled()) {
                    logger.trace("action [{}] completed on shard [{}] for request [{}]", transportReplicaAction, request.shardId(),
                            request);
                }
                setPhase(task, "finished");
                try {
                    channel.sendResponse(replicaResponse);
                } catch (Exception e) {
                    onFailure(e);
                }
            }

            @Override
            public void onFailure(Exception e) {
                responseWithFailure(e);
            }
        }
    }

    private IndexShard getIndexShard(ShardId shardId) {
        IndexService indexService = indicesService.indexServiceSafe(shardId.getIndex());
        return indexService.getShard(shardId.id());
    }

    /**
     * Responsible for routing and retrying failed operations on the primary.
     * The actual primary operation is done in {@link ReplicationOperation} on the
     * node with primary copy.
     *
     * Resolves index and shard id for the request before routing it to target node
     */
    final class ReroutePhase extends AbstractRunnable {
        private final ActionListener<Response> listener;
        private final Request request;
        private final ReplicationTask task;
        private final ClusterStateObserver observer;
        private final AtomicBoolean finished = new AtomicBoolean();

        ReroutePhase(ReplicationTask task, Request request, ActionListener<Response> listener) {
            this.request = request;
            if (task != null) {
                this.request.setParentTask(clusterService.localNode().getId(), task.getId());
            }
            this.listener = listener;
            this.task = task;
            this.observer = new ClusterStateObserver(clusterService, request.timeout(), logger, threadPool.getThreadContext());
        }

        @Override
        public void onFailure(Exception e) {
            finishWithUnexpectedFailure(e);
        }

        @Override
        protected void doRun() {
            setPhase(task, "routing");
            final ClusterState state = observer.observedState();
            if (handleBlockExceptions(state)) {
                return;
            }

            // request does not have a shardId yet, we need to pass the concrete index to resolve shardId
            final String concreteIndex = concreteIndex(state);
            final IndexMetaData indexMetaData = state.metaData().index(concreteIndex);
            if (indexMetaData == null) {
                retry(new IndexNotFoundException(concreteIndex));
                return;
            }

            // resolve all derived request fields, so we can route and apply it
            resolveRequest(state.metaData(), indexMetaData, request);
            assert request.shardId() != null : "request shardId must be set in resolveRequest";
            assert request.waitForActiveShards() != ActiveShardCount.DEFAULT : "request waitForActiveShards must be set in resolveRequest";

            final ShardRouting primary = primary(state);
            if (retryIfUnavailable(state, primary)) {
                return;
            }
            final DiscoveryNode node = state.nodes().get(primary.currentNodeId());
            taskManager.registerChildTask(task, node.getId());
            if (primary.currentNodeId().equals(state.nodes().getLocalNodeId())) {
                performLocalAction(state, primary, node);
            } else {
                performRemoteAction(state, primary, node);
            }
        }

        private void performLocalAction(ClusterState state, ShardRouting primary, DiscoveryNode node) {
            setPhase(task, "waiting_on_primary");
            if (logger.isTraceEnabled()) {
                logger.trace("send action [{}] on primary [{}] for request [{}] with cluster state version [{}] to [{}] ",
                    transportPrimaryAction, request.shardId(), request, state.version(), primary.currentNodeId());
            }
            performAction(node, transportPrimaryAction, true, new ConcreteShardRequest<>(request, primary.allocationId().getId()));
        }

        private void performRemoteAction(ClusterState state, ShardRouting primary, DiscoveryNode node) {
            if (state.version() < request.routedBasedOnClusterVersion()) {
                logger.trace("failed to find primary [{}] for request [{}] despite sender thinking it would be here. Local cluster state "
                        + "version [{}]] is older than on sending node (version [{}]), scheduling a retry...", request.shardId(), request,
                    state.version(), request.routedBasedOnClusterVersion());
                retryBecauseUnavailable(request.shardId(), "failed to find primary as current cluster state with version ["
                    + state.version() + "] is stale (expected at least [" + request.routedBasedOnClusterVersion() + "]");
                return;
            } else {
                // chasing the node with the active primary for a second hop requires that we are at least up-to-date with the current
                // cluster state version this prevents redirect loops between two nodes when a primary was relocated and the relocation
                // target is not aware that it is the active primary shard already.
                request.routedBasedOnClusterVersion(state.version());
            }
            if (logger.isTraceEnabled()) {
                logger.trace("send action [{}] on primary [{}] for request [{}] with cluster state version [{}] to [{}]", actionName,
                    request.shardId(), request, state.version(), primary.currentNodeId());
            }
            setPhase(task, "rerouted");
            performAction(node, actionName, false, request);
        }

        private boolean retryIfUnavailable(ClusterState state, ShardRouting primary) {
            if (primary == null || primary.active() == false) {
                logger.trace("primary shard [{}] is not yet active, scheduling a retry: action [{}], request [{}], "
                    + "cluster state version [{}]", request.shardId(), actionName, request, state.version());
                retryBecauseUnavailable(request.shardId(), "primary shard is not active");
                return true;
            }
            if (state.nodes().nodeExists(primary.currentNodeId()) == false) {
                logger.trace("primary shard [{}] is assigned to an unknown node [{}], scheduling a retry: action [{}], request [{}], "
                    + "cluster state version [{}]", request.shardId(), primary.currentNodeId(), actionName, request, state.version());
                retryBecauseUnavailable(request.shardId(), "primary shard isn't assigned to a known node.");
                return true;
            }
            return false;
        }

        private String concreteIndex(ClusterState state) {
            return resolveIndex() ? indexNameExpressionResolver.concreteSingleIndex(state, request).getName() : request.index();
        }

        private ShardRouting primary(ClusterState state) {
            IndexShardRoutingTable indexShard = state.getRoutingTable().shardRoutingTable(request.shardId());
            return indexShard.primaryShard();
        }

        private boolean handleBlockExceptions(ClusterState state) {
            ClusterBlockException blockException = state.blocks().globalBlockedException(globalBlockLevel());
            if (blockException != null) {
                handleBlockException(blockException);
                return true;
            }
            blockException = state.blocks().indexBlockedException(indexBlockLevel(), concreteIndex(state));
            if (blockException != null) {
                handleBlockException(blockException);
                return true;
            }
            return false;
        }

        private void handleBlockException(ClusterBlockException blockException) {
            if (blockException.retryable()) {
                logger.trace("cluster is blocked, scheduling a retry", blockException);
                retry(blockException);
            } else {
                finishAsFailed(blockException);
            }
        }

        private void performAction(final DiscoveryNode node, final String action, final boolean isPrimaryAction,
                                   final TransportRequest requestToPerform) {
            transportService.sendRequest(node, action, requestToPerform, transportOptions, new TransportResponseHandler<Response>() {

                @Override
                public Response newInstance() {
                    return newResponseInstance();
                }

                @Override
                public String executor() {
                    return ThreadPool.Names.SAME;
                }

                @Override
                public void handleResponse(Response response) {
                    finishOnSuccess(response);
                }

                @Override
                public void handleException(TransportException exp) {
                    try {
                        // if we got disconnected from the node, or the node / shard is not in the right state (being closed)
                        final Throwable cause = exp.unwrapCause();
                        if (cause instanceof ConnectTransportException || cause instanceof NodeClosedException ||
                            (isPrimaryAction && retryPrimaryException(cause))) {
                            logger.trace(
                                (org.apache.logging.log4j.util.Supplier<?>) () -> new ParameterizedMessage(
                                    "received an error from node [{}] for request [{}], scheduling a retry",
                                    node.getId(),
                                    requestToPerform),
                                exp);
                            retry(exp);
                        } else {
                            finishAsFailed(exp);
                        }
                    } catch (Exception e) {
                        e.addSuppressed(exp);
                        finishWithUnexpectedFailure(e);
                    }
                }
            });
        }

        void retry(Exception failure) {
            assert failure != null;
            if (observer.isTimedOut()) {
                // we running as a last attempt after a timeout has happened. don't retry
                finishAsFailed(failure);
                return;
            }
            setPhase(task, "waiting_for_retry");
            request.onRetry();
            final ThreadContext.StoredContext context = threadPool.getThreadContext().newStoredContext();
            observer.waitForNextChange(new ClusterStateObserver.Listener() {
                @Override
                public void onNewClusterState(ClusterState state) {
                    context.close();
                    run();
                }

                @Override
                public void onClusterServiceClose() {
                    finishAsFailed(new NodeClosedException(clusterService.localNode()));
                }

                @Override
                public void onTimeout(TimeValue timeout) {
                    context.close();
                    // Try one more time...
                    run();
                }
            });
        }

        void finishAsFailed(Exception failure) {
            if (finished.compareAndSet(false, true)) {
                setPhase(task, "failed");
                logger.trace(
                    (org.apache.logging.log4j.util.Supplier<?>)
                        () -> new ParameterizedMessage("operation failed. action [{}], request [{}]", actionName, request), failure);
                listener.onFailure(failure);
            } else {
                assert false : "finishAsFailed called but operation is already finished";
            }
        }

        void finishWithUnexpectedFailure(Exception failure) {
            logger.warn(
                (org.apache.logging.log4j.util.Supplier<?>)
                    () -> new ParameterizedMessage(
                        "unexpected error during the primary phase for action [{}], request [{}]",
                        actionName,
                        request),
                failure);
            if (finished.compareAndSet(false, true)) {
                setPhase(task, "failed");
                listener.onFailure(failure);
            } else {
                assert false : "finishWithUnexpectedFailure called but operation is already finished";
            }
        }

        void finishOnSuccess(Response response) {
            if (finished.compareAndSet(false, true)) {
                setPhase(task, "finished");
                if (logger.isTraceEnabled()) {
                    logger.trace("operation succeeded. action [{}],request [{}]", actionName, request);
                }
                listener.onResponse(response);
            } else {
                assert false : "finishOnSuccess called but operation is already finished";
            }
        }

        void retryBecauseUnavailable(ShardId shardId, String message) {
            retry(new UnavailableShardsException(shardId, "{} Timeout: [{}], request: [{}]", message, request.timeout(), request));
        }
    }

    /**
     * Tries to acquire reference to {@link IndexShard} to perform a primary operation. Released after performing primary operation locally
     * and replication of the operation to all replica shards is completed / failed (see {@link ReplicationOperation}).
     */
    private void acquirePrimaryShardReference(ShardId shardId, String allocationId,
                                              ActionListener<PrimaryShardReference> onReferenceAcquired) {
        IndexShard indexShard = getIndexShard(shardId);
        // we may end up here if the cluster state used to route the primary is so stale that the underlying
        // index shard was replaced with a replica. For example - in a two node cluster, if the primary fails
        // the replica will take over and a replica will be assigned to the first node.
        if (indexShard.routingEntry().primary() == false) {
            throw new ReplicationOperation.RetryOnPrimaryException(indexShard.shardId(),
                "actual shard is not a primary " + indexShard.routingEntry());
        }
        final String actualAllocationId = indexShard.routingEntry().allocationId().getId();
        if (actualAllocationId.equals(allocationId) == false) {
            throw new ShardNotFoundException(shardId, "expected aID [{}] but found [{}]", allocationId, actualAllocationId);
        }

        ActionListener<Releasable> onAcquired = new ActionListener<Releasable>() {
            @Override
            public void onResponse(Releasable releasable) {
                onReferenceAcquired.onResponse(new PrimaryShardReference(indexShard, releasable));
            }

            @Override
            public void onFailure(Exception e) {
                onReferenceAcquired.onFailure(e);
            }
        };

        indexShard.acquirePrimaryOperationLock(onAcquired, executor);
    }

    /**
     * Indicated whether this operation should be replicated to shadow replicas or not. If this method returns true the replication phase
     * will be skipped. For example writes such as index and delete don't need to be replicated on shadow replicas but refresh and flush do.
     */
    protected boolean shouldExecuteReplication(Settings settings) {
        return IndexMetaData.isIndexUsingShadowReplicas(settings) == false;
    }

    class ShardReference implements Releasable {

        protected final IndexShard indexShard;
        private final Releasable operationLock;

        ShardReference(IndexShard indexShard, Releasable operationLock) {
            this.indexShard = indexShard;
            this.operationLock = operationLock;
        }

        @Override
        public void close() {
            operationLock.close();
        }

        public long getLocalCheckpoint() {
            return indexShard.getLocalCheckpoint();
        }

        public ShardRouting routingEntry() {
            return indexShard.routingEntry();
        }

    }

    class PrimaryShardReference extends ShardReference implements ReplicationOperation.Primary<Request, ReplicaRequest, PrimaryResult> {

        PrimaryShardReference(IndexShard indexShard, Releasable operationLock) {
            super(indexShard, operationLock);
        }

        public boolean isRelocated() {
            return indexShard.state() == IndexShardState.RELOCATED;
        }

        @Override
        public void failShard(String reason, Exception e) {
            try {
                indexShard.failShard(reason, e);
            } catch (Exception inner) {
                e.addSuppressed(inner);
            }
        }

        @Override
        public PrimaryResult perform(Request request) throws Exception {
            PrimaryResult result = shardOperationOnPrimary(request, indexShard);
            if (result.replicaRequest() != null) {
                result.replicaRequest().primaryTerm(indexShard.getPrimaryTerm());
            }
            return result;
        }

        @Override
        public void updateLocalCheckpointForShard(String allocationId, long checkpoint) {
            indexShard.updateLocalCheckpointForShard(allocationId, checkpoint);
        }

        @Override
        public long localCheckpoint() {
            return indexShard.getLocalCheckpoint();
        }

    }

    public static class ReplicaResponse extends ActionResponse implements ReplicationOperation.ReplicaResponse {
        private long localCheckpoint;
        private String allocationId;

        ReplicaResponse() {

        }

        public ReplicaResponse(String allocationId, long localCheckpoint) {
            this.allocationId = allocationId;
            this.localCheckpoint = localCheckpoint;
        }

        @Override
        public void readFrom(StreamInput in) throws IOException {
            super.readFrom(in);
            localCheckpoint = in.readZLong();
            allocationId = in.readString();
        }

        @Override
        public void writeTo(StreamOutput out) throws IOException {
            super.writeTo(out);
            out.writeZLong(localCheckpoint);
            out.writeString(allocationId);
        }

        @Override
        public long localCheckpoint() {
            return localCheckpoint;
        }

        @Override
        public String allocationId() {
            return allocationId;
        }
    }

    final class ReplicasProxy implements ReplicationOperation.Replicas<ReplicaRequest> {

        @Override
        public void performOn(ShardRouting replica, ReplicaRequest request, ActionListener<ReplicationOperation.ReplicaResponse> listener) {
            String nodeId = replica.currentNodeId();
            final DiscoveryNode node = clusterService.state().nodes().get(nodeId);
            if (node == null) {
                listener.onFailure(new NoNodeAvailableException("unknown node [" + nodeId + "]"));
                return;
            }
            transportService.sendRequest(node, transportReplicaAction,
                new ConcreteShardRequest<>(request, replica.allocationId().getId()), transportOptions,
                // Eclipse can't handle when this is <> so we specify the type here.
                new ActionListenerResponseHandler<ReplicaResponse>(listener, ReplicaResponse::new));
        }

        @Override
        public void failShard(ShardRouting replica, long primaryTerm, String message, Exception exception,
                              Runnable onSuccess, Consumer<Exception> onPrimaryDemoted, Consumer<Exception> onIgnoredFailure) {
            shardStateAction.remoteShardFailed(replica.shardId(), replica.allocationId().getId(), primaryTerm, message, exception,
                createListener(onSuccess, onPrimaryDemoted, onIgnoredFailure));
        }

        @Override
        public void markShardCopyAsStale(ShardId shardId, String allocationId, long primaryTerm, Runnable onSuccess,
                                         Consumer<Exception> onPrimaryDemoted, Consumer<Exception> onIgnoredFailure) {
            shardStateAction.remoteShardFailed(shardId, allocationId, primaryTerm, "mark copy as stale", null,
                createListener(onSuccess, onPrimaryDemoted, onIgnoredFailure));
        }

        private ShardStateAction.Listener createListener(final Runnable onSuccess, final Consumer<Exception> onPrimaryDemoted,
                                                         final Consumer<Exception> onIgnoredFailure) {
            return new ShardStateAction.Listener() {
                @Override
                public void onSuccess() {
                    onSuccess.run();
                }

                @Override
                public void onFailure(Exception shardFailedError) {
                    if (shardFailedError instanceof ShardStateAction.NoLongerPrimaryShardException) {
                        onPrimaryDemoted.accept(shardFailedError);
                    } else {
                        // these can occur if the node is shutting down and are okay
                        // any other exception here is not expected and merits investigation
                        assert shardFailedError instanceof TransportException ||
                            shardFailedError instanceof NodeClosedException : shardFailedError;
                        onIgnoredFailure.accept(shardFailedError);
                    }
                }
            };
        }
    }

    /** a wrapper class to encapsulate a request when being sent to a specific allocation id **/
    public static final class ConcreteShardRequest<R extends TransportRequest> extends TransportRequest {

        /** {@link AllocationId#getId()} of the shard this request is sent to **/
        private String targetAllocationID;

        private R request;

        ConcreteShardRequest(Supplier<R> requestSupplier) {
            request = requestSupplier.get();
            // null now, but will be populated by reading from the streams
            targetAllocationID = null;
        }

        ConcreteShardRequest(R request, String targetAllocationID) {
            Objects.requireNonNull(request);
            Objects.requireNonNull(targetAllocationID);
            this.request = request;
            this.targetAllocationID = targetAllocationID;
        }

        @Override
        public void setParentTask(String parentTaskNode, long parentTaskId) {
            request.setParentTask(parentTaskNode, parentTaskId);
        }

        @Override
        public void setParentTask(TaskId taskId) {
            request.setParentTask(taskId);
        }

        @Override
        public TaskId getParentTask() {
            return request.getParentTask();
        }
        @Override
        public Task createTask(long id, String type, String action, TaskId parentTaskId) {
            return request.createTask(id, type, action, parentTaskId);
        }

        @Override
        public String getDescription() {
            return "[" + request.getDescription() + "] for aID [" + targetAllocationID + "]";
        }

        @Override
        public void readFrom(StreamInput in) throws IOException {
            targetAllocationID = in.readString();
            request.readFrom(in);
        }

        @Override
        public void writeTo(StreamOutput out) throws IOException {
            out.writeString(targetAllocationID);
            request.writeTo(out);
        }

        public R getRequest() {
            return request;
        }

        public String getTargetAllocationID() {
            return targetAllocationID;
        }

        @Override
        public String toString() {
            return "request: " + request + ", target allocation id: " + targetAllocationID;
        }
    }

    /**
     * Sets the current phase on the task if it isn't null. Pulled into its own
     * method because its more convenient that way.
     */
    static void setPhase(ReplicationTask task, String phase) {
        if (task != null) {
            task.setPhase(phase);
        }
    }
}


text:
"""
Async helper function that are invalid syntax on Python 3.5 and below.

This code is best effort, and may have edge cases not behaving as expected. In
particular it contain a number of heuristics to detect whether code is
effectively async and need to run in an event loop or not.

Some constructs (like top-level `return`, or `yield`) are taken care of
explicitly to actually raise a SyntaxError and stay as close as possible to
Python semantics.
"""

import ast
import sys
from textwrap import dedent, indent

class _AsyncIORunner:

    def __call__(self, coro):
        """
        Handler for asyncio autoawait
        """
        import asyncio

        return asyncio.get_event_loop().run_until_complete(coro)

    def __str__(self):
        return 'asyncio'

_asyncio_runner = _AsyncIORunner()

def _curio_runner(coroutine):
    """
    handler for curio autoawait
    """
    import curio

    return curio.run(coroutine)

def _trio_runner(async_fn):
    import trio

    async def loc(coro):
        """
        We need the dummy no-op async def to protect from
        trio's internal. See https://github.com/python-trio/trio/issues/89
        """
        return await coro

    return trio.run(loc, async_fn)

def _pseudo_sync_runner(coro):
    """
    A runner that does not really allow async execution, and just advance the coroutine.

    See discussion in https://github.com/python-trio/trio/issues/608,

    Credit to Nathaniel Smith

    """
    try:
        coro.send(None)
    except StopIteration as exc:
        return exc.value
    else:
        # TODO: do not raise but return an execution result with the right info.
        raise RuntimeError(
            "{coro_name!r} needs a real async loop".format(coro_name=coro.__name__)
        )

def _asyncify(code: str) -> str:
    """wrap code in async def definition.

    And setup a bit of context to run it later.
    """
    res = dedent(
        """
    async def __wrapper__():
        try:
    {usercode}
        finally:
            locals()
    """
    ).format(usercode=indent(code, " " * 8))
    return res

class _AsyncSyntaxErrorVisitor(ast.NodeVisitor):
    """
    Find syntax errors that would be an error in an async repl, but because
    the implementation involves wrapping the repl in an async function, it
    is erroneously allowed (e.g. yield or return at the top level)
    """
    def __init__(self):
        self.depth = 0
        super().__init__()

    def generic_visit(self, node):
        func_types = (ast.FunctionDef, ast.AsyncFunctionDef)
        invalid_types_by_depth = {
            0: (ast.Return, ast.Yield, ast.YieldFrom),
            1: (ast.Nonlocal,)
        }

        should_traverse = self.depth < max(invalid_types_by_depth.keys())
        if isinstance(node, func_types) and should_traverse:
            self.depth += 1
            super().generic_visit(node)
            self.depth -= 1
        elif isinstance(node, invalid_types_by_depth[self.depth]):
            raise SyntaxError()
        else:
            super().generic_visit(node)

def _async_parse_cell(cell: str) -> ast.AST:
    """
    This is a compatibility shim for pre-3.7 when async outside of a function
    is a syntax error at the parse stage.

    It will return an abstract syntax tree parsed as if async and await outside
    of a function were not a syntax error.
    """
    if sys.version_info < (3, 7):
        # Prior to 3.7 you need to asyncify before parse
        wrapped_parse_tree = ast.parse(_asyncify(cell))
        return wrapped_parse_tree.body[0].body[0]
    else:
        return ast.parse(cell)

def _should_be_async(cell: str) -> bool:
    """Detect if a block of code need to be wrapped in an `async def`

    Attempt to parse the block of code, it it compile we're fine.
    Otherwise we  wrap if and try to compile.

    If it works, assume it should be async. Otherwise Return False.

    Not handled yet: If the block of code has a return statement as  the top
    level, it will be seen as async. This is a know limitation.
    """

    try:
        # we can't limit ourself to ast.parse, as it __accepts__ to parse on
        # 3.7+, but just does not _compile_
        compile(cell, "<>", "exec")
        return False
    except SyntaxError:
        try:
            parse_tree = _async_parse_cell(cell)

            # Raise a SyntaxError if there are top-level return or yields
            v = _AsyncSyntaxErrorVisitor()
            v.visit(parse_tree)

        except SyntaxError:
            return False
        return True
    return False

redpajama_set_name: RedPajamaGithub


text:
using System.Text.Json;
using Azure.Core;

namespace Azure.Graph.Rbac.Models
{
    public partial class ApplicationBase : IUtf8JsonSerializable
    {
        void IUtf8JsonSerializable.Write(Utf8JsonWriter writer)
        {
            writer.WriteStartObject();
            if (Optional.IsDefined(AllowGuestsSignIn))
            {
                writer.WritePropertyName("allowGuestsSignIn");
                writer.WriteBooleanValue(AllowGuestsSignIn.Value);
            }
            if (Optional.IsDefined(AllowPassthroughUsers))
            {
                writer.WritePropertyName("allowPassthroughUsers");
                writer.WriteBooleanValue(AllowPassthroughUsers.Value);
            }
            if (Optional.IsDefined(AppLogoUrl))
            {
                writer.WritePropertyName("appLogoUrl");
                writer.WriteStringValue(AppLogoUrl);
            }
            if (Optional.IsCollectionDefined(AppRoles))
            {
                writer.WritePropertyName("appRoles");
                writer.WriteStartArray();
                foreach (var item in AppRoles)
                {
                    writer.WriteObjectValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsCollectionDefined(AppPermissions))
            {
                writer.WritePropertyName("appPermissions");
                writer.WriteStartArray();
                foreach (var item in AppPermissions)
                {
                    writer.WriteStringValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsDefined(AvailableToOtherTenants))
            {
                writer.WritePropertyName("availableToOtherTenants");
                writer.WriteBooleanValue(AvailableToOtherTenants.Value);
            }
            if (Optional.IsDefined(ErrorUrl))
            {
                writer.WritePropertyName("errorUrl");
                writer.WriteStringValue(ErrorUrl);
            }
            if (Optional.IsDefined(GroupMembershipClaims))
            {
                writer.WritePropertyName("groupMembershipClaims");
                writer.WriteStringValue(GroupMembershipClaims.Value.ToString());
            }
            if (Optional.IsDefined(Homepage))
            {
                writer.WritePropertyName("homepage");
                writer.WriteStringValue(Homepage);
            }
            if (Optional.IsDefined(InformationalUrls))
            {
                writer.WritePropertyName("informationalUrls");
                writer.WriteObjectValue(InformationalUrls);
            }
            if (Optional.IsDefined(IsDeviceOnlyAuthSupported))
            {
                writer.WritePropertyName("isDeviceOnlyAuthSupported");
                writer.WriteBooleanValue(IsDeviceOnlyAuthSupported.Value);
            }
            if (Optional.IsCollectionDefined(KeyCredentials))
            {
                writer.WritePropertyName("keyCredentials");
                writer.WriteStartArray();
                foreach (var item in KeyCredentials)
                {
                    writer.WriteObjectValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsCollectionDefined(KnownClientApplications))
            {
                writer.WritePropertyName("knownClientApplications");
                writer.WriteStartArray();
                foreach (var item in KnownClientApplications)
                {
                    writer.WriteStringValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsDefined(LogoutUrl))
            {
                writer.WritePropertyName("logoutUrl");
                writer.WriteStringValue(LogoutUrl);
            }
            if (Optional.IsDefined(Oauth2AllowImplicitFlow))
            {
                writer.WritePropertyName("oauth2AllowImplicitFlow");
                writer.WriteBooleanValue(Oauth2AllowImplicitFlow.Value);
            }
            if (Optional.IsDefined(Oauth2AllowUrlPathMatching))
            {
                writer.WritePropertyName("oauth2AllowUrlPathMatching");
                writer.WriteBooleanValue(Oauth2AllowUrlPathMatching.Value);
            }
            if (Optional.IsCollectionDefined(Oauth2Permissions))
            {
                writer.WritePropertyName("oauth2Permissions");
                writer.WriteStartArray();
                foreach (var item in Oauth2Permissions)
                {
                    writer.WriteObjectValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsDefined(Oauth2RequirePostResponse))
            {
                writer.WritePropertyName("oauth2RequirePostResponse");
                writer.WriteBooleanValue(Oauth2RequirePostResponse.Value);
            }
            if (Optional.IsCollectionDefined(OrgRestrictions))
            {
                writer.WritePropertyName("orgRestrictions");
                writer.WriteStartArray();
                foreach (var item in OrgRestrictions)
                {
                    writer.WriteStringValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsDefined(OptionalClaims))
            {
                writer.WritePropertyName("optionalClaims");
                writer.WriteObjectValue(OptionalClaims);
            }
            if (Optional.IsCollectionDefined(PasswordCredentials))
            {
                writer.WritePropertyName("passwordCredentials");
                writer.WriteStartArray();
                foreach (var item in PasswordCredentials)
                {
                    writer.WriteObjectValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsCollectionDefined(PreAuthorizedApplications))
            {
                writer.WritePropertyName("preAuthorizedApplications");
                writer.WriteStartArray();
                foreach (var item in PreAuthorizedApplications)
                {
                    writer.WriteObjectValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsDefined(PublicClient))
            {
                writer.WritePropertyName("publicClient");
                writer.WriteBooleanValue(PublicClient.Value);
            }
            if (Optional.IsDefined(PublisherDomain))
            {
                writer.WritePropertyName("publisherDomain");
                writer.WriteStringValue(PublisherDomain);
            }
            if (Optional.IsCollectionDefined(ReplyUrls))
            {
                writer.WritePropertyName("replyUrls");
                writer.WriteStartArray();
                foreach (var item in ReplyUrls)
                {
                    writer.WriteStringValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsCollectionDefined(RequiredResourceAccess))
            {
                writer.WritePropertyName("requiredResourceAccess");
                writer.WriteStartArray();
                foreach (var item in RequiredResourceAccess)
                {
                    writer.WriteObjectValue(item);
                }
                writer.WriteEndArray();
            }
            if (Optional.IsDefined(SamlMetadataUrl))
            {
                writer.WritePropertyName("samlMetadataUrl");
                writer.WriteStringValue(SamlMetadataUrl);
            }
            if (Optional.IsDefined(SignInAudience))
            {
                writer.WritePropertyName("signInAudience");
                writer.WriteStringValue(SignInAudience);
            }
            if (Optional.IsDefined(WwwHomepage))
            {
                writer.WritePropertyName("wwwHomepage");
                writer.WriteStringValue(WwwHomepage);
            }
            writer.WriteEndObject();
        }
    }
}


text:
package org.krugler.yalder.text;

import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.io.IOUtils;
import org.krugler.yalder.BaseLanguageModel;
import org.krugler.yalder.LanguageLocale;

/**
 * Encapsulation of model about a given language. This consists of
 * a list of ngrams, with a (normalized) count for each ngram.
 *
 * @author Ken Krugler
 *
 */

public class TextLanguageModel extends BaseLanguageModel {

    // Constants used when writing/reading the model.
    private static final String VERSION_PREFIX = "version:";
    private static final String LANGUAGE_PREFIX = "language:";
    private static final String NGRAM_SIZE_PREFIX = "max ngram length:";
    private static final String NGRAM_DATA_PREFIX = "ngrams:";

    // Map from ngram to count
    private Map<String, Integer> _normalizedCounts;

    /**
     * No-arg construct for deserialization
     */
    public TextLanguageModel() {
        super();
    }

    public TextLanguageModel(LanguageLocale modelLanguage, int maxNGramLength, Map<String, Integer> normalizedCounts) {
        super(modelLanguage, maxNGramLength);

        _normalizedCounts = normalizedCounts;
    }

    public int size() {
        return _normalizedCounts.size();
    }

    public int getNGramCount(String ngram) {
        Integer result = _normalizedCounts.get(ngram);
        return result == null ? 0 : result;
    }

    public Map<String, Integer> getNGramCounts() {
        return _normalizedCounts;
    }

    @Override
    public int prune(int minNormalizedCount) {
        Set<String> ngramsToPrune = new HashSet<>();
        for (String ngram : _normalizedCounts.keySet()) {
            if (_normalizedCounts.get(ngram) < minNormalizedCount) {
                ngramsToPrune.add(ngram);
            }
        }

        for (String ngram : ngramsToPrune) {
            _normalizedCounts.remove(ngram);
        }

        return ngramsToPrune.size();
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = super.hashCode();
        result = prime * result + ((_normalizedCounts == null) ? 0 : _normalizedCounts.hashCode());
        return result;
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj)
            return true;
        if (!super.equals(obj))
            return false;
        if (getClass() != obj.getClass())
            return false;
        TextLanguageModel other = (TextLanguageModel) obj;
        if (_normalizedCounts == null) {
            if (other._normalizedCounts != null)
                return false;
        } else if (!_normalizedCounts.equals(other._normalizedCounts))
            return false;
        return true;
    }

    public void writeAsText(OutputStreamWriter osw) throws IOException {
        osw.write(String.format("%s %d\n", VERSION_PREFIX, MODEL_VERSION));
        osw.write(String.format("%s %s\n", LANGUAGE_PREFIX, _modelLanguage.getName()));
        osw.write(String.format("%s %d\n", NGRAM_SIZE_PREFIX, _maxNGramLength));
        osw.write(String.format("%s\n", NGRAM_DATA_PREFIX));

        for (String ngram : _normalizedCounts.keySet()) {
            osw.write(String.format("\t%s: %d\n", ngram, _normalizedCounts.get(ngram)));
        }
    }

    public void readAsText(InputStreamReader isw) throws IOException {
        List<String> lines = IOUtils.readLines(isw);
        if (lines.size() < 5) {
            throw new IllegalArgumentException("Model doesn't contain enough lines of text");
        }

        // First line must be the version
        String versionLine = lines.get(0);
        if (!versionLine.startsWith(VERSION_PREFIX)) {
            throw new IllegalArgumentException("First line of model must be the version number");
        }

        versionLine = versionLine.substring(VERSION_PREFIX.length()).trim();
        int version = Integer.parseInt(versionLine);
        if (version != MODEL_VERSION) {
            throw new IllegalArgumentException("Version doesn't match supported values, got " + version);
        }

        // Next line must be language info.
        String languageLine = lines.get(1);
        if (!languageLine.startsWith(LANGUAGE_PREFIX)) {
            throw new IllegalArgumentException("Second line of model must be the language info");
        }

        languageLine = languageLine.substring(LANGUAGE_PREFIX.length()).trim();
        _modelLanguage = LanguageLocale.fromString(languageLine);

        // Next line is the ngram max length
        String ngramSizeLine = lines.get(2);
        if (!ngramSizeLine.startsWith(NGRAM_SIZE_PREFIX)) {
            throw new IllegalArgumentException("Third line of model must be the max ngram length");
        }

        ngramSizeLine = ngramSizeLine.substring(NGRAM_SIZE_PREFIX.length()).trim();
        _maxNGramLength = Integer.parseInt(ngramSizeLine);

        // Next line is the ngram header
        String ngramsLine = lines.get(3);
        if (!ngramsLine.equals(NGRAM_DATA_PREFIX)) {
            throw new IllegalArgumentException("Fourth line of model must be the ngram data header");
        }

        Pattern p = Pattern.compile("\t(.+?): (.+)");
        Matcher m = p.matcher("");

        _normalizedCounts = new HashMap<String, Integer>(lines.size() - 4);
        for (int i = 4; i < lines.size(); i++) {
            String ngramLine = lines.get(i);
            m.reset(ngramLine);
            if (!m.matches()) {
                throw new IllegalArgumentException(String.format("%d ngram in model has invalid format", i - 3));
            }

            String ngram = m.group(1);
            int normalizedCount = Integer.parseInt(m.group(2));
            _normalizedCounts.put(ngram, normalizedCount);
        }
    }

}


text:
"""Lint checks used by all the linters."""

from __future__ import annotations

import os
import re

from . import js_ts_linter
from . import warranted_angular_security_bypasses

from .. import build
from .. import common
from .. import concurrent_task_utils

EXCLUDED_PATHS = (
    'third_party/*', 'build/*', '.git/*', '*.pyc', 'CHANGELOG',
    'integrations/*', 'integrations_dev/*', '*.svg', '*.gif', '*.png',
    '*.webp', '*.zip', '*.ico', '*.jpg', '*.min.js', 'backend_prod_files/*',
    'assets/scripts/*', 'core/domain/proto/*.py', 'core/tests/data/*',
    'core/tests/build_sources/*', '*.mp3', '*.mp4', 'node_modules/*',
    'typings/*', 'local_compiled_js/*', 'webpack_bundles/*',
    'core/tests/services_sources/*', 'core/tests/release_sources/tmp_unzip.zip',
    'scripts/linters/test_files/*', 'proto_files/*',
    'core/tests/release_sources/tmp_unzip.tar.gz',
    'core/templates/combined-tests.spec.ts',
    'core/templates/css/oppia-material.css',
    'core/templates/google-analytics.initializer.ts',
    'extensions/classifiers/proto/*',
    '%s/*' % js_ts_linter.COMPILED_TYPESCRIPT_TMP_PATH)

GENERATED_FILE_PATHS = (
    'core/templates/expressions/parser.js',)

CONFIG_FILE_PATHS = (
    'core/tests/.browserstack.env.example',
    'core/tests/protractor.conf.js',
    'core/tests/karma.conf.ts',
    'core/templates/mathjaxConfig.ts',
    'assets/constants.ts',
    'assets/rich_text_components_definitions.ts',
    'webpack.config.ts',
    'webpack.dev.config.ts',
    'webpack.prod.config.ts')

BAD_STRINGS_CONSTANTS = {
    '"DEV_MODE": false': {
        'message': 'Please set the DEV_MODE variable in constants.ts '
                   'to true before committing.',
        'excluded_files': ()
    },
    '"EMULATOR_MODE": false': {
        'message': 'Please set the EMULATOR_MODE variable in constants.ts '
                   'to true before committing.',
        'excluded_files': ()
    }
}

BAD_PATTERNS = {
    '\t': {
        'message': 'Please use spaces instead of tabs.',
        'excluded_files': (),
        'excluded_dirs': (
            'assets/i18n/', 'core/tests/build_sources/assets/')},
    '\r': {
        'message': 'Please make sure all files only have LF endings (no CRLF).',
        'excluded_files': (),
        'excluded_dirs': ()},
    '<<<<<<<': {
        'message': 'Please fully resolve existing merge conflicts.',
        'excluded_files': (),
        'excluded_dirs': ()},
    '>>>>>>>': {
        'message': 'Please fully resolve existing merge conflicts.',
        'excluded_files': (),
        'excluded_dirs': ()},
    'glyphicon': {
        'message': 'Please use equivalent material-icons '
                   'instead of glyphicons.',
        'excluded_files': (),
        'excluded_dirs': ()}
}

BAD_PATTERNS_REGEXP = [
    {
        'regexp': re.compile(r'TODO[^\(]*[^\)][^:]*[^A-Z]+[^\w]*$'),
        'message': 'Please link TODO comments to an issue '
                   'in the format TODO(#issuenum): XXX. ',
        'excluded_files': (),
        'excluded_dirs': ()
    }
]

MANDATORY_PATTERNS_REGEXP = [
    {
        'regexp': re.compile(
            r'Copyright \d{4} The Oppia Authors\. All Rights Reserved\.'),
        'message': 'Please ensure this file should contain a proper '
                   'copyright notice.',
        'included_types': ('.py', '.js', '.sh', '.ts'),
        'excluded_files': GENERATED_FILE_PATHS + CONFIG_FILE_PATHS + (
            '__init__.py', ),
        'excluded_dirs': EXCLUDED_PATHS
    },
    {
        'regexp': re.compile('from __future__ import annotations'),
        'message': 'Please ensure this file should contain annotations '
                   'future import.',
        'included_types': ('.py'),
        'excluded_files': GENERATED_FILE_PATHS + CONFIG_FILE_PATHS + (
            '__init__.py',),
        'excluded_dirs': EXCLUDED_PATHS
    }
]

MANDATORY_PATTERNS_JS_REGEXP = [
    {
        'regexp': re.compile(r'^\s\*\s@fileoverview\s[a-zA-Z0-9_]+'),
        'message': 'Please ensure this file should contain a file '
                   'overview i.e. a short description of the file.',
        'included_types': ('.js', '.ts'),
        'excluded_files': GENERATED_FILE_PATHS + CONFIG_FILE_PATHS,
        'excluded_dirs': EXCLUDED_PATHS
    }
]

BAD_LINE_PATTERNS_HTML_REGEXP = [
    {
        'regexp': re.compile(r'text\/ng-template'),
        'message': 'The directives must be directly referenced.',
        'excluded_files': (),
        'excluded_dirs': (
            'extensions/answer_summarizers/',
            'extensions/classifiers/',
            'extensions/objects/',
            'extensions/value_generators/')
    },
    {
        'regexp': re.compile(r'[ \t]+$'),
        'message': 'There should not be any trailing whitespaces.',
        'excluded_files': (),
        'excluded_dirs': ()
    },
    {
        'regexp': re.compile(r'\$parent'),
        'message': 'Please do not access parent properties ' +
                   'using $parent. Use the scope object ' +
                   'for this purpose.',
        'excluded_files': (),
        'excluded_dirs': ()
    },
    {
        'regexp': re.compile(r'\s+style\s*=\s*'),
        'message': 'Please do not use inline styling.',
        'excluded_files': (),
        'excluded_dirs': ()
    }
]

BAD_PATTERNS_PYTHON_REGEXP = [
    {
        'regexp': re.compile(r'__author__'),
        'message': 'Please remove author tags from this file.',
        'excluded_files': (),
        'excluded_dirs': ()
    },
    {
        'regexp': re.compile(r'ndb\.'),
        'message': (
            'Please use datastore_services instead of ndb, for example:\n'
            '\n'
            'datastore_services = models.Registry.import_datastore_services()\n'
            '\n'
            'class SampleModel(datastore_services.Model):\n'
            '    ...\n'),
        'excluded_files': (),
        'excluded_dirs': ('core/platform',),
    },
    {
        'regexp': re.compile(r'\Wprint\('),
        'message': 'Please do not use print statement.',
        'excluded_files': (
            'core/tests/test_utils.py',
            'core/tests/performance_framework/perf_domain.py',
            'core/tests/test_utils_test.py'),
        'excluded_dirs': ('scripts/',)
    },
    {
        'regexp': re.compile(r'# pylint:\s*disable=[A-Z][0-9]{4}'),
        'message': 'Please remove pylint exclusion if it is unnecessary, or '
                   'make it human readable with a sentence instead of an id. '
                   'The id-to-message list can be seen '
                   'here->http://pylint-messages.wikidot.com/all-codes',
        'excluded_files': (),
        'excluded_dirs': ()
    },
]

BAD_PATTERNS_MAP = {
    '.html': BAD_LINE_PATTERNS_HTML_REGEXP,
    '.py': BAD_PATTERNS_PYTHON_REGEXP
}

def is_filepath_excluded_for_bad_patterns_check(pattern, filepath):
    """Checks if file is excluded from the bad patterns check.

    Args:
        pattern: str. The pattern to be checked against.
        filepath: str. Path of the file.

    Returns:
        bool. Whether to exclude the given file from this
        particular pattern check.
    """
    return (any(
        filepath.startswith(bad_pattern)
        for bad_pattern in BAD_PATTERNS[pattern]['excluded_dirs'])
            or filepath in BAD_PATTERNS[pattern]['excluded_files'])

def check_bad_pattern_in_file(filepath, file_content, pattern):
    """Detects whether the given pattern is present in the file.

    Args:
        filepath: str. Path of the file.
        file_content: str. Contents of the file.
        pattern: dict. (regexp(regex pattern) : Object containing details for
            the pattern to be checked. Pattern to match:
                message: str. Message to show if pattern matches.
                excluded_files: tuple(str). Files to be excluded from matching.
                excluded_dirs: tuple(str). Directories to be excluded from
                    matching).

    Returns:
        tuple(bool, list(str)). A 2-tuple whose first element is a bool
        which set to True if there is bad pattern found else False, whose second
        element is a list of failed messages.
    """
    error_messages = []
    failed = False
    regexp = pattern['regexp']
    if not (any(
            filepath.startswith(excluded_dir)
            for excluded_dir in pattern['excluded_dirs'])
            or any(
                filepath.endswith(excluded_file)
                for excluded_file in pattern['excluded_files'])):
        bad_pattern_count = 0
        for line_num, line in enumerate(file_content, 1):
            if line.endswith('\n'):
                stripped_line = line[:-1]
            else:
                stripped_line = line
            if stripped_line.endswith('disable-bad-pattern-check'):
                continue
            if regexp.search(stripped_line):
                error_message = ('%s --> Line %s: %s' % (
                    filepath, line_num, pattern['message']))
                error_messages.append(error_message)
                bad_pattern_count += 1
        if bad_pattern_count:
            failed = True
            return failed, error_messages
    return failed, error_messages

def check_file_type_specific_bad_pattern(filepath, content):
    """Check the file content based on the file's extension.

    Args:
        filepath: str. Path of the file.
        content: str. Contents of the file.

    Returns:
        bool. True if there is bad pattern else false.
        total_error_count: int. The number of errors.
    """
    error_messages = []
    failed = False
    _, extension = os.path.splitext(filepath)
    pattern = BAD_PATTERNS_MAP.get(extension)
    total_error_count = 0
    if pattern:
        for regexp in pattern:
            failed, error_message = check_bad_pattern_in_file(
                filepath, content, regexp)
            error_messages.extend(error_message)
            if failed:
                total_error_count += 1
    if total_error_count:
        failed = True
    return failed, total_error_count, error_messages

class GeneralPurposeLinter:
    """Manages all the common linting functions. As an abstract base class, this
    is not intended to be used directly.
    """

    def __init__(self, files_to_lint, file_cache):
        """Constructs a GeneralPurposeLinter object.

        Args:
            files_to_lint: list(str). A list of filepaths to lint.
            file_cache: object(FileCache). Provides thread-safe access to cached
                file content.
        """
        # Set path for node.
        # The path for node is set explicitly, since otherwise the lint
        # tests fail on CircleCI due to the TypeScript files not being
        # compilable.
        os.environ['PATH'] = '%s/bin:' % common.NODE_PATH + os.environ['PATH']

        self.files_to_lint = files_to_lint
        self.file_cache = file_cache

    @property
    def all_filepaths(self):
        """Returns all file paths."""
        return self.files_to_lint

    def _check_for_mandatory_pattern_in_file(
            self, pattern_list, filepath, failed):
        """Checks for a given mandatory pattern in a file.

        Args:
            pattern_list: list(dict). The list of the mandatory patterns list to
                be checked for in the file.
            filepath: str. The path to the file to be linted.
            failed: bool. Status of failure of the check.

        Returns:
            bool. The failure status of the check.
        """
        # This boolean list keeps track of the regex matches
        # found in the file.
        pattern_found_list = []
        error_messages = []

        try:
            file_content = self.file_cache.readlines(filepath)
        except Exception as e:
            raise Exception('%s %s' % (filepath, e))
        for index, regexp_to_check in enumerate(
                pattern_list):
            if (any(filepath.endswith(
                    allowed_type) for allowed_type in (
                        regexp_to_check['included_types'])) and (
                            not any(
                                filepath.endswith(
                                    pattern) for pattern in (
                                        regexp_to_check['excluded_files'] +
                                        regexp_to_check['excluded_dirs'])))):
                pattern_found_list.append(index)
                for line in file_content:
                    if regexp_to_check['regexp'].search(line):
                        pattern_found_list.pop()
                        break
        if pattern_found_list:
            failed = True
            for pattern_found in pattern_found_list:
                error_message = ('%s --> %s' % (
                    filepath,
                    pattern_list[pattern_found]['message']))
                error_messages.append(error_message)

        return failed, error_messages

    def check_mandatory_patterns(self):
        """This function checks that all files contain the mandatory
        patterns.
        """
        name = 'Mandatory pattern'
        error_messages = []
        failed = False
        sets_of_patterns_to_match = [
            MANDATORY_PATTERNS_REGEXP, MANDATORY_PATTERNS_JS_REGEXP]
        for filepath in self.all_filepaths:
            for pattern_list in sets_of_patterns_to_match:
                failed, mandatory_error_messages = (
                    self._check_for_mandatory_pattern_in_file(
                        pattern_list, filepath, failed))
                error_messages.extend(mandatory_error_messages)
        return concurrent_task_utils.TaskResult(
            name, failed, error_messages, error_messages)

    def check_bad_patterns(self):
        """This function is used for detecting bad patterns."""
        name = 'Bad pattern'
        total_files_checked = 0
        total_error_count = 0
        error_messages = []
        all_filepaths = [
            filepath for filepath in self.all_filepaths if not (
                filepath.endswith('general_purpose_linter.py') or (
                    filepath.endswith('general_purpose_linter_test.py')))]
        failed = False
        for filepath in all_filepaths:
            file_content = self.file_cache.readlines(filepath)
            total_files_checked += 1
            for pattern, error in BAD_PATTERNS.items():
                if is_filepath_excluded_for_bad_patterns_check(
                        pattern, filepath):
                    continue
                for line_num, line in enumerate(file_content):
                    if pattern in line:
                        failed = True
                        error_message = ('%s --> Line %s: %s' % (
                            filepath, line_num + 1,
                            error['message']))
                        error_messages.append(error_message)
                        total_error_count += 1

            for regexp in BAD_PATTERNS_REGEXP:
                bad_pattern_check_failed, bad_pattern_error_messages = (
                    check_bad_pattern_in_file(
                        filepath, file_content, regexp))
                if bad_pattern_check_failed:
                    error_messages.extend(bad_pattern_error_messages)
                    total_error_count += 1

            (
                file_type_specific_bad_pattern_failed,
                temp_count, bad_pattern_error_messages) = (
                    check_file_type_specific_bad_pattern(
                        filepath, file_content))
            failed = (
                failed or file_type_specific_bad_pattern_failed or
                bad_pattern_check_failed)
            total_error_count += temp_count
            error_messages.extend(bad_pattern_error_messages)

            if filepath.endswith('constants.ts'):
                for pattern, constants in BAD_STRINGS_CONSTANTS.items():
                    for line in file_content:
                        if pattern in line:
                            failed = True
                            error_message = ('%s --> %s' % (
                                filepath,
                                constants['message']))
                            error_messages.append(error_message)
                            total_error_count += 1
        return concurrent_task_utils.TaskResult(
            name, failed, error_messages, error_messages)

    def check_newline_at_eof(self):
        """This function is used to detect newline at the end of file."""
        name = 'Newline at EOF'
        error_messages = []
        files_to_lint = self.all_filepaths
        failed = False

        for filepath in files_to_lint:
            file_content = self.file_cache.readlines(filepath)
            file_length = len(file_content)
            if (
                    file_length >= 1 and
                    not re.search(r'[^\n]\n', file_content[-1])):
                error_message = (
                    '%s --> There should be a single newline at the '
                    'end of file.' % filepath)
                error_messages.append(error_message)
                failed = True
        return concurrent_task_utils.TaskResult(
            name, failed, error_messages, error_messages)

    def check_disallowed_flags(self):
        """This function is used to disallow flags."""
        name = 'Disallow flags'
        disallow_flag = (
            'eslint-disable-next-line oppia/no-bypass-security-phrase')
        error_messages = []
        files_to_lint = self.all_filepaths
        failed = False
        excluded_files = (
            warranted_angular_security_bypasses
            .EXCLUDED_BYPASS_SECURITY_TRUST_FILES)
        allowed_files = ''
        for filepath in files_to_lint:
            for excluded_file in excluded_files:
                if excluded_file in filepath:
                    allowed_files = filepath
            if not filepath.endswith('.ts') or filepath == allowed_files:
                continue
            file_content = self.file_cache.read(filepath)

            if disallow_flag in file_content:
                error_message = (
                    '%s --> Please do not use "no-bypass-security-phrase" flag.'
                    ' It is only expected to be used in files listed in'
                    ' warranted_angular_security_bypasses.py' % filepath)
                error_messages.append(error_message)
                failed = True
        return concurrent_task_utils.TaskResult(
            name, failed, error_messages, error_messages)

    def check_extra_js_files(self):
        """Checks if the changes made include extra js files in core
        or extensions folder which are not specified in
        build.JS_FILEPATHS_NOT_TO_BUILD.

        Returns:
            TaskResult. A TaskResult object representing the result of the lint
            check.
        """
        name = 'Extra JS files'
        error_messages = []
        files_to_lint = self.all_filepaths
        failed = False

        for filepath in files_to_lint:
            if filepath.endswith(
                ('.js')) and filepath.startswith(
                    ('core/templates', 'extensions')) and (
                        filepath not in build.JS_FILEPATHS_NOT_TO_BUILD
                        ) and not filepath.endswith('protractor.js'):
                error_message = (
                    '%s  --> Found extra .js file' % filepath)
                error_messages.append(error_message)
                failed = True

        if failed:
            err_msg = (
                'If you want the above files to be present as js files, '
                'add them to the list JS_FILEPATHS_NOT_TO_BUILD in '
                'build.py. Otherwise, rename them to .ts')
            error_messages.append(err_msg)
        return concurrent_task_utils.TaskResult(
            name, failed, error_messages, error_messages)

    def perform_all_lint_checks(self):
        """Perform all the lint checks and returns the messages returned by all
        the checks.

        Returns:
            list(TaskResult). A list of TaskResult objects representing the
            results of the lint checks.
        """
        if not self.all_filepaths:
            return [
                concurrent_task_utils.TaskResult(
                    'General purpose lint', False, [],
                    ['There are no files to be checked.'])]
        task_results = [
            self.check_mandatory_patterns(), self.check_bad_patterns(),
            self.check_newline_at_eof(), self.check_extra_js_files(),
            self.check_disallowed_flags()]
        return task_results

def get_linters(files_to_lint, file_cache):
    """Creates GeneralPurposeLinter object and returns it.

    Args:
        files_to_lint: list(str). A list of filepaths to lint.
        file_cache: object(FileCache). Provides thread-safe access to cached
            file content.

    Returns:
        tuple(GeneralPurposeLinter, None). A 2-tuple of custom and third_party
        linter objects.
    """
    custom_linter = GeneralPurposeLinter(files_to_lint, file_cache)

    return custom_linter, None


text:
module Control.Semigroupoid where

class Semigroupoid c where
  (.) ::
    c y z
    -> c x y
    -> c x z


text:
#ifndef ATILDEFS_H
#include "AtilDefs.h"
#endif

#ifndef STRINGBUFFER_H
#define STRINGBUFFER_H
#if  __GNUC__ >= 4
#pragma GCC visibility push(default)
#endif

namespace Atil
{

///
/// The <c>StringBuffer</c> class is ATIL's class for avoiding string handling.
/// It forces all strings into a PASCAL form of byte count and array of bytes
/// model of string handling. It includes no facilities for concatenation nor
/// parsing. No searching is implemented. The purpose of the class is primarily
/// to carry strings from the application to the format codecs and vice versa.
///
///
class StringBuffer
{
public:
    ///
    /// This enum lists the character types that <c>StringBuffer</c> can wrap.
    /// It is up the the "user" to interpret the bytes contained as a string
    /// of one enum's type.
    ///
    ///
    enum CharacterType {
        ///  This entry indicates a specific char encoding scheme.
        kASCII,
        ///  This entry indicates a specific char encoding scheme.
        kUTF_8,
        ///  This entry indicates a specific char encoding scheme.
        kUTF_16,
        ///  This entry indicates a specific char encoding scheme.
        kUTF_32,
        ///  This entry indicates a specific char encoding scheme.
        kMBCS
    };

    ///
    /// This static method will calculate the number of bytes in a string
    /// of the given <c>CharacterType</c>.
    ///
    ///
    /// <param name="pString">This is a pointer to an array of bytes that
    /// holds a string.
    /// </param>
    ///
    /// <param name="type">This is a value of the <c>StringBuffer::CharacterType</c>
    /// that describes the character type of the passed in string.
    /// </param>
    ///
    /// <returns>
    /// This method returns the number of bytes needed to store the string including
    /// any needed termination character.
    /// </returns>
    ///
    static int bytesInString ( const Byte* pString, CharacterType type );

    ///
    /// This method will attempt to convert from one character type to another.
    ///
    ///
    /// <param name="sbString">The <c>StringBuffer</c> instance to be converted.
    /// </param>
    ///
    /// <param name="sbResult">A reference to a <c>StringBuffer</c> that will be set
    /// to the converted result.
    /// </param>
    ///
    /// <param name="type">The <c>CharacterType</c> that the converted string should have.
    /// </param>
    ///
    /// <remarks>
    /// The conversion offered by this method is basic and uses the default conversion
    /// methods offered by the host operating systems. As such, it serves for file path
    /// conversions but is not appropriate for more sophisticated text.
    /// </remarks>
    ///
    static void convertString ( const StringBuffer& sbString,
        StringBuffer& sbResult, CharacterType type );

    ///
    /// This returns <c>CharacterType</c> under which ATIL was built.
    ///
    ///
	/// <returns>
	/// Returns a member of the <c>CharacterType</c> enum.
	/// </returns>
	///
    static CharacterType defaultCharacterType ();

public:

    ///
    /// The default (empty) constructor.
    ///
    ///
    StringBuffer ();

    ///
    /// The initialization constructor.
    ///
    ///
    /// <param name="nNumBytes">The number bytes to allocate to store the string.
    /// </param>
    ///
    /// <param name="type">The <c>CharacterType</c> of the string that will be stored.
    /// </param>
    ///
    StringBuffer (int nNumBytes, CharacterType type = StringBuffer::kASCII);

    ///
    /// The initialization constructor.
    ///
    ///
    /// <param name="nNumBytes">The number bytes to allocate to store the string.
    /// </param>
    ///
    /// <param name="pBytes">A const pointer to a buffer containing the number of bytes
    /// of data passed in that are to be stored.
    /// </param>
    ///
    /// <param name="type">The <c>CharacterType</c> of the string that will be stored.
    /// </param>
    ///
    StringBuffer (int nNumBytes, const Byte* pBytes, CharacterType type = StringBuffer::kASCII);

    ///
    /// The copy constructor.
    ///
    ///
	/// <param name="x">
	/// A const reference to the <c>StringBuffer</c> to copy.
	/// </param>
	///
    StringBuffer (const StringBuffer& x);

    ///
    /// The destructor.
    ///
    ///
    virtual ~StringBuffer ();

    ///
    /// This method returns a const pointer to the internal byte array that holds the string.
    ///
    ///
	/// <returns>
	/// Returns a const pointer to a <c>Byte</c> array.
	/// </returns>
	///
    virtual const Byte* data () const;

    ///
    /// This method will return the size in bytes of the internal byte
    /// array that holds the string.
    ///
    ///
	/// <returns>
	/// Returns the integer number of bytes held in the internal array.
	/// </returns>
	///
    virtual int numDataBytes () const;

    ///
    /// This method returns the <c>CharacterType</c> that has been set as the encoding
    /// type of the internal byte array that holds the string.
    ///
    ///
	/// <returns>
	/// Returns a member of the <c>CharacterType</c> enum.
	/// </returns>
	///
    virtual CharacterType type () const;

    ///
    /// Like the assignment operator this method will over write the object with the values
    /// of the argument.
    ///
    ///
	/// <returns>
	/// Returns a reference to <c>*this</c>.
	/// </returns>
	///
    virtual StringBuffer& set (const StringBuffer& string);

    ///
    /// This method mimics the initialization constructor and will over write the current
    /// object members with the arguments.
    ///
    ///
    /// <param name="nNumBytes">The number bytes to allocate to store the string.
    /// </param>
    ///
    /// <param name="pChar">A const pointer to a buffer containing the number of bytes
    /// of data passed in that are to be stored.
    /// </param>
    ///
    /// <param name="type">The <c>CharacterType</c> of the string that will be stored.
    /// </param>
    ///
	/// <returns>
	/// Returns a reference to <c>*this</c>.
	/// </returns>
	///
    virtual StringBuffer& set (int numBytes, const Byte* pChar, CharacterType type = StringBuffer::kASCII);

    ///
    /// This is a simplistic test that will test the all the bytes of the string
    /// and return true if they are all zero.
    ///
	///
	/// <returns>
	/// Returns true if the first internal array is empty.
	/// </returns>
	///
    virtual bool isZero () const;

    ///
    /// The assignment operator will cause the current values to be over written with those
    /// of the argument.
    ///
    ///
	/// <param name="rhs">
	/// The argument to be copied.
	///</param>
	///
	/// <returns>
	/// Returns a reference to <c>*this</c>.
	/// </returns>
	///
    StringBuffer& operator=( const StringBuffer& rhs );

    ///
    /// The equals operator will return true if all the members of each are equal respectively.
    /// The test for equivalence is a byte wise test for each byte of the embedded buffer.
    ///
    ///
	/// <param name="rhs">
	/// The argument to be compared.
	///</param>
	///
	/// <returns>
	/// Returns true if the instance are equivalent.
	/// </returns>
	///
    bool operator==( const StringBuffer& rhs ) const;

    ///
    /// The not equals operator will return true if any member does not equal that of the
    /// argument.
    ///
    ///
	/// <param name="rhs">
	/// The argument to be compared.
	///</param>
	///
	/// <returns>
	/// Returns true if the instance are not equivalent.
	/// </returns>
	///
    bool operator!=( const StringBuffer& rhs ) const;

    ///  Cast operator to convert to a <c>Atil::char_t</c> for use as the
    /// default string type on the platform ATIL is compiled for.
    /// <returns> This returns a <c>const char_t*</c> pointer for use as a string.</returns>
    operator const char_t*() const;
private:
    int mNumBytes;
    Byte* mpData;
    CharacterType  meCharType;
};

} //end of namespace
#if  __GNUC__ >= 4
#pragma GCC visibility pop
#endif

#endif