text:
package org.apache.ignite.internal.processors.hadoop.shuffle;

import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.processors.hadoop.HadoopJobEx;
import org.apache.ignite.internal.processors.hadoop.HadoopJobId;
import org.apache.ignite.internal.processors.hadoop.HadoopMapperAwareTaskOutput;
import org.apache.ignite.internal.processors.hadoop.HadoopMapperUtils;
import org.apache.ignite.internal.processors.hadoop.HadoopPartitioner;
import org.apache.ignite.internal.processors.hadoop.HadoopSerialization;
import org.apache.ignite.internal.processors.hadoop.HadoopTaskContext;
import org.apache.ignite.internal.processors.hadoop.HadoopTaskInfo;
import org.apache.ignite.internal.processors.hadoop.HadoopTaskInput;
import org.apache.ignite.internal.processors.hadoop.HadoopTaskOutput;
import org.apache.ignite.internal.processors.hadoop.HadoopTaskType;
import org.apache.ignite.internal.processors.hadoop.counter.HadoopPerformanceCounter;
import org.apache.ignite.internal.processors.hadoop.message.HadoopMessage;
import org.apache.ignite.internal.processors.hadoop.shuffle.collections.HadoopConcurrentHashMultimap;
import org.apache.ignite.internal.processors.hadoop.shuffle.collections.HadoopMultimap;
import org.apache.ignite.internal.processors.hadoop.shuffle.collections.HadoopSkipList;
import org.apache.ignite.internal.processors.hadoop.shuffle.direct.HadoopDirectDataInput;
import org.apache.ignite.internal.processors.hadoop.shuffle.direct.HadoopDirectDataOutputContext;
import org.apache.ignite.internal.processors.hadoop.shuffle.direct.HadoopDirectDataOutputState;
import org.apache.ignite.internal.util.GridUnsafe;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.io.GridUnsafeDataInput;
import org.apache.ignite.internal.util.lang.GridClosureException;
import org.apache.ignite.internal.util.lang.IgniteInClosure2X;
import org.apache.ignite.internal.util.offheap.unsafe.GridUnsafeMemory;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.thread.IgniteThread;
import org.jetbrains.annotations.Nullable;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.zip.GZIPInputStream;

import static org.apache.ignite.internal.processors.hadoop.HadoopJobProperty.PARTITION_HASHMAP_SIZE;
import static org.apache.ignite.internal.processors.hadoop.HadoopJobProperty.SHUFFLE_JOB_THROTTLE;
import static org.apache.ignite.internal.processors.hadoop.HadoopJobProperty.SHUFFLE_MAPPER_STRIPED_OUTPUT;
import static org.apache.ignite.internal.processors.hadoop.HadoopJobProperty.SHUFFLE_MSG_GZIP;
import static org.apache.ignite.internal.processors.hadoop.HadoopJobProperty.SHUFFLE_MSG_SIZE;
import static org.apache.ignite.internal.processors.hadoop.HadoopJobProperty.SHUFFLE_REDUCER_NO_SORTING;
import static org.apache.ignite.internal.processors.hadoop.HadoopJobProperty.get;

/**
 * Shuffle job.
 */
public class HadoopShuffleJob<T> implements AutoCloseable {
    /** */
    private static final int DFLT_SHUFFLE_MSG_SIZE = 1024 * 1024;

    /** */
    private static final boolean DFLT_SHUFFLE_MSG_GZIP = false;

    /** */
    private final HadoopJobEx job;

    /** */
    private final GridUnsafeMemory mem;

    /** */
    private final boolean needPartitioner;

    /** Task contexts for each reduce task. */
    private final AtomicReferenceArray<LocalTaskContextProxy> locReducersCtx;

    /** Reducers addresses. */
    private T[] reduceAddrs;

    /** Total reducer count. */
    private final int totalReducerCnt;

    /** Local reducers address. */
    private final T locReduceAddr;

    /** */
    private final HadoopShuffleMessage[] msgs;

    /** Maps for local reducers. */
    private final AtomicReferenceArray<HadoopMultimap> locMaps;

    /** Maps for remote reducers. */
    private final AtomicReferenceArray<HadoopMultimap> rmtMaps;

    /** */
    private volatile IgniteInClosure2X<T, HadoopMessage> io;

    /** */
    protected ConcurrentMap<Long, IgniteBiTuple<HadoopShuffleMessage, GridFutureAdapter<?>>> sentMsgs =
        new ConcurrentHashMap<>();

    /** */
    private volatile GridWorker snd;

    /** Latch for remote addresses waiting. */
    private final CountDownLatch ioInitLatch = new CountDownLatch(1);

    /** Finished flag. Set on flush or close. */
    private volatile boolean flushed;

    /** */
    private final IgniteLogger log;

    /** Message size. */
    private final int msgSize;

    /** Whether to GZIP shuffle messages. */
    private final boolean msgGzip;

    /** Whether to strip mappers for remote execution. */
    private final boolean stripeMappers;

    /** Local shuffle states. */
    private volatile HashMap<T, HadoopShuffleLocalState> locShuffleStates = new HashMap<>();

    /** Remote shuffle states. */
    private volatile HashMap<T, HadoopShuffleRemoteState> rmtShuffleStates = new HashMap<>();

    /** Mutex for internal synchronization. */
    private final Object mux = new Object();

    /** */
    private final long throttle;

    /** Embedded mode flag. */
    private final boolean embedded;

    /**
     * @param locReduceAddr Local reducer address.
     * @param log Logger.
     * @param job Job.
     * @param mem Memory.
     * @param totalReducerCnt Amount of reducers in the Job.
     * @param locReducers Reducers will work on current node.
     * @param locMappersCnt Number of mappers running on the given node.
     * @param embedded Whether shuffle is running in embedded mode.
     * @throws IgniteCheckedException If error.
     */
    public HadoopShuffleJob(T locReduceAddr, IgniteLogger log, HadoopJobEx job, GridUnsafeMemory mem,
        int totalReducerCnt, int[] locReducers, int locMappersCnt, boolean embedded) throws IgniteCheckedException {
        this.locReduceAddr = locReduceAddr;
        this.totalReducerCnt = totalReducerCnt;
        this.job = job;
        this.mem = mem;
        this.log = log.getLogger(HadoopShuffleJob.class);
        this.embedded = embedded;

        boolean stripeMappers0 = get(job.info(), SHUFFLE_MAPPER_STRIPED_OUTPUT, true);

        if (stripeMappers0) {
            if (!embedded) {
                log.info("Striped mapper output is disabled becuase it cannot be used in external mode [jobId=" +
                    job.id() + ']');

                stripeMappers0 = false;
            }
        }

        stripeMappers = stripeMappers0;

        msgSize = get(job.info(), SHUFFLE_MSG_SIZE, DFLT_SHUFFLE_MSG_SIZE);
        msgGzip = get(job.info(), SHUFFLE_MSG_GZIP, DFLT_SHUFFLE_MSG_GZIP);

        locReducersCtx = new AtomicReferenceArray<>(totalReducerCnt);

        if (!F.isEmpty(locReducers)) {
            for (int rdc : locReducers) {
                HadoopTaskInfo taskInfo = new HadoopTaskInfo(HadoopTaskType.REDUCE, job.id(), rdc, 0, null);

                locReducersCtx.set(rdc, new LocalTaskContextProxy(taskInfo));
            }
        }

        needPartitioner = totalReducerCnt > 1;

        // Size of local map is always equal to total reducer number to allow index-based lookup.
        locMaps = new AtomicReferenceArray<>(totalReducerCnt);

        // Size of remote map:
        // - If there are no local mappers, then we will not send anything, so set to 0;
        // - If output is not striped, then match it to total reducer count, the same way as for local maps.
        // - If output is striped, then multiply previous value by number of local mappers.
        int rmtMapsSize = locMappersCnt == 0 ? 0 : totalReducerCnt;

        if (stripeMappers)
            rmtMapsSize *= locMappersCnt;

        rmtMaps = new AtomicReferenceArray<>(rmtMapsSize);
        msgs = new HadoopShuffleMessage[rmtMapsSize];

        throttle = get(job.info(), SHUFFLE_JOB_THROTTLE, 0);
    }

    /**
     * @param reduceAddrs Addresses of reducers.
     * @return {@code True} if addresses were initialized by this call.
     */
    public boolean initializeReduceAddresses(T[] reduceAddrs) {
        if (this.reduceAddrs == null) {
            this.reduceAddrs = reduceAddrs;

            return true;
        }

        return false;
    }

    /**
     * @return {@code True} if reducers addresses were initialized.
     */
    public boolean reducersInitialized() {
        return reduceAddrs != null;
    }

    /**
     * @param igniteInstanceName Ignite instance name.
     * @param io IO Closure for sending messages.
     */
    @SuppressWarnings("BusyWait")
    public void startSending(String igniteInstanceName, IgniteInClosure2X<T, HadoopMessage> io) {
        assert snd == null;
        assert io != null;

        this.io = io;

        if (!stripeMappers) {
            if (!flushed) {
                snd = new GridWorker(igniteInstanceName, "hadoop-shuffle-" + job.id(), log) {
                    @Override protected void body() throws InterruptedException {
                        try {
                            while (!isCancelled()) {
                                if (throttle > 0)
                                    Thread.sleep(throttle);

                                collectUpdatesAndSend(false);
                            }
                        }
                        catch (IgniteCheckedException e) {
                            throw new IllegalStateException(e);
                        }
                    }
                };

                new IgniteThread(snd).start();
            }
        }

        ioInitLatch.countDown();
    }

    /**
     * @param maps Maps.
     * @param idx Index.
     * @return Map.
     */
    private HadoopMultimap getOrCreateMap(AtomicReferenceArray<HadoopMultimap> maps, int idx) {
        HadoopMultimap map = maps.get(idx);

        if (map == null) { // Create new map.
            map = get(job.info(), SHUFFLE_REDUCER_NO_SORTING, false) ?
                new HadoopConcurrentHashMultimap(job.info(), mem, get(job.info(), PARTITION_HASHMAP_SIZE, 8 * 1024)):
                new HadoopSkipList(job.info(), mem);

            if (!maps.compareAndSet(idx, null, map)) {
                map.close();

                return maps.get(idx);
            }
        }

        return map;
    }

    /**
     * @param src Source.
     * @param msg Message.
     * @throws IgniteCheckedException Exception.
     */
    public void onShuffleMessage(T src, HadoopShuffleMessage msg) throws IgniteCheckedException {
        assert msg.buffer() != null;
        assert msg.offset() > 0;

        HadoopTaskContext taskCtx = locReducersCtx.get(msg.reducer()).get();

        HadoopPerformanceCounter perfCntr = HadoopPerformanceCounter.getCounter(taskCtx.counters(), null);

        perfCntr.onShuffleMessage(msg.reducer(), U.currentTimeMillis());

        HadoopMultimap map = getOrCreateMap(locMaps, msg.reducer());

        // Add data from message to the map.
        try (HadoopMultimap.Adder adder = map.startAdding(taskCtx)) {
            final GridUnsafeDataInput dataInput = new GridUnsafeDataInput();
            final UnsafeValue val = new UnsafeValue(msg.buffer());

            msg.visit(new HadoopShuffleMessage.Visitor() {
                /** */
                private HadoopMultimap.Key key;

                @Override public void onKey(byte[] buf, int off, int len) throws IgniteCheckedException {
                    dataInput.bytes(buf, off, off + len);

                    key = adder.addKey(dataInput, key);
                }

                @Override public void onValue(byte[] buf, int off, int len) {
                    val.off = off;
                    val.size = len;

                    key.add(val);
                }
            });
        }

        if (embedded) {
            // No immediate response.
            if (localShuffleState(src).onShuffleMessage())
                sendFinishResponse(src, msg.jobId());
        }
        else
            // Response for every message.
            io.apply(src, new HadoopShuffleAck(msg.id(), msg.jobId()));
    }

    /**
     * Process shuffle message.
     *
     * @param src Source.
     * @param msg Message.
     * @throws IgniteCheckedException Exception.
     */
    public void onDirectShuffleMessage(T src, HadoopDirectShuffleMessage msg) throws IgniteCheckedException {
        byte[] buf = extractBuffer(msg);

        assert buf != null;

        int rdc = msg.reducer();

        HadoopTaskContext taskCtx = locReducersCtx.get(rdc).get();

        HadoopPerformanceCounter perfCntr = HadoopPerformanceCounter.getCounter(taskCtx.counters(), null);

        perfCntr.onShuffleMessage(rdc, U.currentTimeMillis());

        HadoopMultimap map = getOrCreateMap(locMaps, rdc);

        HadoopSerialization keySer = taskCtx.keySerialization();
        HadoopSerialization valSer = taskCtx.valueSerialization();

        // Add data from message to the map.
        try (HadoopMultimap.Adder adder = map.startAdding(taskCtx)) {
            HadoopDirectDataInput in = new HadoopDirectDataInput(buf);

            Object key = null;
            Object val = null;

            for (int i = 0; i < msg.count(); i++) {
                key = keySer.read(in, key);
                val = valSer.read(in, val);

                adder.write(key, val);
            }
        }

        if (localShuffleState(src).onShuffleMessage())
            sendFinishResponse(src, msg.jobId());
    }

    /**
     * Extract buffer from direct shuffle message.
     *
     * @param msg Message.
     * @return Buffer.
     * @throws IgniteCheckedException On error.
     */
    private byte[] extractBuffer(HadoopDirectShuffleMessage msg) throws IgniteCheckedException {
        if (msgGzip) {
            byte[] res = new byte[msg.dataLength()];

            try (GZIPInputStream in = new GZIPInputStream(new ByteArrayInputStream(msg.buffer()), res.length)) {
                int len = in.read(res, 0, res.length);

                assert len == res.length;
            }
            catch (IOException e) {
                throw new IgniteCheckedException("Failed to uncompress direct shuffle message.", e);
            }

            return res;
        }
        else
            return msg.buffer();
    }

    /**
     * @param ack Shuffle ack.
     */
    @SuppressWarnings("ConstantConditions")
    public void onShuffleAck(HadoopShuffleAck ack) {
        IgniteBiTuple<HadoopShuffleMessage, GridFutureAdapter<?>> tup = sentMsgs.get(ack.id());

        if (tup != null)
            tup.get2().onDone();
        else
            log.warning("Received shuffle ack for not registered shuffle id: " + ack);
    }

    /**
     * Process shuffle finish request.
     *
     * @param src Source.
     * @param msg Shuffle finish message.
     */
    public void onShuffleFinishRequest(T src, HadoopShuffleFinishRequest msg) {
        if (log.isDebugEnabled())
            log.debug("Received shuffle finish request [jobId=" + job.id() + ", src=" + src + ", req=" + msg + ']');

        HadoopShuffleLocalState state = localShuffleState(src);

        if (state.onShuffleFinishMessage(msg.messageCount()))
            sendFinishResponse(src, msg.jobId());
    }

    /**
     * Process shuffle finish response.
     *
     * @param src Source.
     */
    public void onShuffleFinishResponse(T src) {
        if (log.isDebugEnabled())
            log.debug("Received shuffle finish response [jobId=" + job.id() + ", src=" + src + ']');

        remoteShuffleState(src).onShuffleFinishResponse();
    }

    /**
     * Send finish response.
     *
     * @param dest Destination.
     * @param jobId Job ID.
     */
    @SuppressWarnings("unchecked")
    private void sendFinishResponse(T dest, HadoopJobId jobId) {
        if (log.isDebugEnabled())
            log.debug("Sent shuffle finish response [jobId=" + jobId + ", dest=" + dest + ']');

        HadoopShuffleFinishResponse msg = new HadoopShuffleFinishResponse(jobId);

        io.apply(dest, msg);
    }

    /**
     * Get local shuffle state for node.
     *
     * @param src Source
     * @return Local shuffle state.
     */
    private HadoopShuffleLocalState localShuffleState(T src) {
        HashMap<T, HadoopShuffleLocalState> states = locShuffleStates;

        HadoopShuffleLocalState res = states.get(src);

        if (res == null) {
            synchronized (mux) {
                res = locShuffleStates.get(src);

                if (res == null) {
                    res = new HadoopShuffleLocalState();

                    states = new HashMap<>(locShuffleStates);

                    states.put(src, res);

                    locShuffleStates = states;
                }
            }
        }

        return res;
    }

    /**
     * Get remote shuffle state for node.
     *
     * @param src Source.
     * @return Remote shuffle state.
     */
    private HadoopShuffleRemoteState remoteShuffleState(T src) {
        HashMap<T, HadoopShuffleRemoteState> states = rmtShuffleStates;

        HadoopShuffleRemoteState res = states.get(src);

        if (res == null) {
            synchronized (mux) {
                res = rmtShuffleStates.get(src);

                if (res == null) {
                    res = new HadoopShuffleRemoteState();

                    states = new HashMap<>(rmtShuffleStates);

                    states.put(src, res);

                    rmtShuffleStates = states;
                }
            }
        }

        return res;
    }

    /**
     * Get all remote shuffle states.
     *
     * @return Remote shuffle states.
     */
    private HashMap<T, HadoopShuffleRemoteState> remoteShuffleStates() {
        synchronized (mux) {
            return new HashMap<>(rmtShuffleStates);
        }
    }

    /**
     * Unsafe value.
     */
    private static class UnsafeValue implements HadoopMultimap.Value {
        /** */
        private final byte[] buf;

        /** */
        private int off;

        /** */
        private int size;

        /**
         * @param buf Buffer.
         */
        private UnsafeValue(byte[] buf) {
            assert buf != null;

            this.buf = buf;
        }

        /** */
        @Override public int size() {
            return size;
        }

        /** */
        @Override public void copyTo(long ptr) {
            GridUnsafe.copyHeapOffheap(buf, GridUnsafe.BYTE_ARR_OFF + off, ptr, size);
        }
    }

    /**
     * Send updates to remote reducers.
     *
     * @param flush Flush flag.
     * @throws IgniteCheckedException If failed.
     */
    private void collectUpdatesAndSend(boolean flush) throws IgniteCheckedException {
        for (int i = 0; i < rmtMaps.length(); i++)
            collectUpdatesAndSend(i, flush);
    }

    /**
     * Send updates to concrete remote reducer.
     *
     * @param rmtMapIdx Remote map index.
     * @param flush Flush flag.
     * @throws IgniteCheckedException If failed.
     */
    private void collectUpdatesAndSend(int rmtMapIdx, boolean flush) throws IgniteCheckedException {
        final int rmtRdcIdx = stripeMappers ? rmtMapIdx % totalReducerCnt : rmtMapIdx;

        HadoopMultimap map = rmtMaps.get(rmtMapIdx);

        if (map == null)
            return;

        if (msgs[rmtMapIdx] == null)
            msgs[rmtMapIdx] = new HadoopShuffleMessage(job.id(), rmtRdcIdx, msgSize);

        visit(map, rmtMapIdx, rmtRdcIdx);

        if (flush && msgs[rmtMapIdx].offset() != 0)
            send(rmtMapIdx, rmtRdcIdx, 0);
    }

    /**
     * Flush remote direct context.
     *
     * @param rmtMapIdx Remote map index.
     * @param rmtDirectCtx Remote direct context.
     * @param reset Whether to perform reset.
     */
    private void sendShuffleMessage(int rmtMapIdx, @Nullable HadoopDirectDataOutputContext rmtDirectCtx,
        boolean reset) {
        if (rmtDirectCtx == null)
            return;

        int cnt = rmtDirectCtx.count();

        if (cnt == 0)
            return;

        int rmtRdcIdx = stripeMappers ? rmtMapIdx % totalReducerCnt : rmtMapIdx;

        HadoopDirectDataOutputState state = rmtDirectCtx.state();

        if (reset)
            rmtDirectCtx.reset();

        HadoopDirectShuffleMessage msg = new HadoopDirectShuffleMessage(job.id(), rmtRdcIdx, cnt,
            state.buffer(), state.bufferLength(), state.dataLength());

        T nodeId = reduceAddrs[rmtRdcIdx];

        io.apply(nodeId, msg);

        remoteShuffleState(nodeId).onShuffleMessage();
    }

    /**
     * Visit output map.
     *
     * @param map Map.
     * @param rmtMapIdx Remote map index.
     * @param rmtRdcIdx Remote reducer index.
     * @throws IgniteCheckedException If failed.
     */
    private void visit(HadoopMultimap map, final int rmtMapIdx, final int rmtRdcIdx) throws IgniteCheckedException {
        map.visit(false, new HadoopMultimap.Visitor() {
            /** */
            private long keyPtr;

            /** */
            private int keySize;

            /** */
            private boolean keyAdded;

            /** {@inheritDoc} */
            @Override public void onKey(long keyPtr, int keySize) {
                this.keyPtr = keyPtr;
                this.keySize = keySize;

                keyAdded = false;
            }

            private boolean tryAdd(long valPtr, int valSize) {
                HadoopShuffleMessage msg = msgs[rmtMapIdx];

                if (!keyAdded) { // Add key and value.
                    int size = keySize + valSize;

                    if (!msg.available(size, false))
                        return false;

                    msg.addKey(keyPtr, keySize);
                    msg.addValue(valPtr, valSize);

                    keyAdded = true;

                    return true;
                }

                if (!msg.available(valSize, true))
                    return false;

                msg.addValue(valPtr, valSize);

                return true;
            }

            /** {@inheritDoc} */
            @Override public void onValue(long valPtr, int valSize) {
                if (tryAdd(valPtr, valSize))
                    return;

                send(rmtMapIdx, rmtRdcIdx, keySize + valSize);

                keyAdded = false;

                if (!tryAdd(valPtr, valSize))
                    throw new IllegalStateException();
            }
        });
    }

    /**
     * Send message.
     *
     * @param rmtMapIdx Remote map index.
     * @param rmtRdcIdx Remote reducer index.
     * @param newBufMinSize Min new buffer size.
     */
    private void send(int rmtMapIdx, int rmtRdcIdx, int newBufMinSize) {
        HadoopShuffleMessage msg = msgs[rmtMapIdx];

        final long msgId = msg.id();

        final GridFutureAdapter<?> fut;

        if (embedded)
            fut = null;
        else {
            fut = new GridFutureAdapter<>();

            IgniteBiTuple<HadoopShuffleMessage, GridFutureAdapter<?>> old = sentMsgs.putIfAbsent(msgId,
                new IgniteBiTuple<HadoopShuffleMessage, GridFutureAdapter<?>>(msg, fut));

            assert old == null;
        }

        try {
            io.apply(reduceAddrs[rmtRdcIdx], msg);

            if (embedded)
                remoteShuffleState(reduceAddrs[rmtRdcIdx]).onShuffleMessage();
        }
        catch (GridClosureException e) {
            if (fut != null)
                fut.onDone(U.unwrap(e));
        }

        if (fut != null) {
            fut.listen(new IgniteInClosure<IgniteInternalFuture<?>>() {
                @Override public void apply(IgniteInternalFuture<?> f) {
                    try {
                        f.get();

                        // Clean up the future from map only if there was no exception.
                        // Otherwise flush() should fail.
                        sentMsgs.remove(msgId);
                    }
                    catch (IgniteCheckedException e) {
                        log.error("Failed to send message.", e);
                    }
                }
            });
        }

        msgs[rmtMapIdx] = newBufMinSize == 0 ? null : new HadoopShuffleMessage(job.id(), rmtRdcIdx,
            Math.max(msgSize, newBufMinSize));
    }

    /** {@inheritDoc} */
    @Override public void close() throws IgniteCheckedException {
        if (snd != null) {
            snd.cancel();

            try {
                snd.join();
            }
            catch (InterruptedException e) {
                throw new IgniteInterruptedCheckedException(e);
            }
        }

        close(locMaps);
        close(rmtMaps);
    }

    /**
     * @param maps Maps.
     */
    private void close(AtomicReferenceArray<HadoopMultimap> maps) {
        for (int i = 0; i < maps.length(); i++) {
            HadoopMultimap map = maps.get(i);

            if (map != null)
                map.close();
        }
    }

    /**
     * @return Future.
     */
    @SuppressWarnings("unchecked")
    public IgniteInternalFuture<?> flush() throws IgniteCheckedException {
        if (log.isDebugEnabled())
            log.debug("Flushing job " + job.id() + " on address " + locReduceAddr);

        flushed = true;

        if (totalReducerCnt == 0)
            return new GridFinishedFuture<>();

        if (!stripeMappers) {
            U.await(ioInitLatch);

            GridWorker snd0 = snd;

            if (snd0 != null) {
                if (log.isDebugEnabled())
                    log.debug("Cancelling sender thread.");

                snd0.cancel();

                try {
                    snd0.join();

                    if (log.isDebugEnabled())
                        log.debug("Finished waiting for sending thread to complete on shuffle job flush: " + job.id());
                }
                catch (InterruptedException e) {
                    throw new IgniteInterruptedCheckedException(e);
                }
            }

            collectUpdatesAndSend(true); // With flush.

            if (log.isDebugEnabled())
                log.debug("Finished sending collected updates to remote reducers: " + job.id());
        }

        GridCompoundFuture fut = new GridCompoundFuture<>();

        if (embedded) {
            boolean sent = false;

            for (Map.Entry<T, HadoopShuffleRemoteState> rmtStateEntry : remoteShuffleStates().entrySet()) {
                T dest = rmtStateEntry.getKey();
                HadoopShuffleRemoteState rmtState = rmtStateEntry.getValue();

                HadoopShuffleFinishRequest req = new HadoopShuffleFinishRequest(job.id(), rmtState.messageCount());

                io.apply(dest, req);

                if (log.isDebugEnabled())
                    log.debug("Sent shuffle finish request [jobId=" + job.id() + ", dest=" + dest +
                        ", req=" + req + ']');

                fut.add(rmtState.future());

                sent = true;
            }

            if (sent)
                fut.markInitialized();
            else
                return new GridFinishedFuture<>();
        }
        else {
            for (IgniteBiTuple<HadoopShuffleMessage, GridFutureAdapter<?>> tup : sentMsgs.values())
                fut.add(tup.get2());

            fut.markInitialized();

            if (log.isDebugEnabled())
                log.debug("Collected futures to compound futures for flush: " + sentMsgs.size());
        }

        return fut;
    }

    /**
     * @param taskCtx Task context.
     * @return Output.
     * @throws IgniteCheckedException If failed.
     */
    public HadoopTaskOutput output(HadoopTaskContext taskCtx) throws IgniteCheckedException {
        switch (taskCtx.taskInfo().type()) {
            case MAP:
                assert !job.info().hasCombiner() : "The output creation is allowed if combiner has not been defined.";

            case COMBINE:
                return new PartitionedOutput(taskCtx);

            default:
                throw new IllegalStateException("Illegal type: " + taskCtx.taskInfo().type());
        }
    }

    /**
     * @param taskCtx Task context.
     * @return Input.
     * @throws IgniteCheckedException If failed.
     */
    @SuppressWarnings("unchecked")
    public HadoopTaskInput input(HadoopTaskContext taskCtx) throws IgniteCheckedException {
        switch (taskCtx.taskInfo().type()) {
            case REDUCE:
                int reducer = taskCtx.taskInfo().taskNumber();

                HadoopMultimap m = locMaps.get(reducer);

                if (m != null)
                    return m.input(taskCtx);

                return new HadoopTaskInput() { // Empty input.
                    @Override public boolean next() {
                        return false;
                    }

                    @Override public Object key() {
                        throw new IllegalStateException();
                    }

                    @Override public Iterator<?> values() {
                        throw new IllegalStateException();
                    }

                    @Override public void close() {
                        // No-op.
                    }
                };

            default:
                throw new IllegalStateException("Illegal type: " + taskCtx.taskInfo().type());
        }
    }

    /**
     * Check if certain partition (reducer) is local.
     *
     * @param part Partition.
     * @return {@code True} if local.
     */
    private boolean isLocalPartition(int part) {
        return locReducersCtx.get(part) != null;
    }

    /**
     * Partitioned output.
     */
    public class PartitionedOutput implements HadoopMapperAwareTaskOutput {
        /** */
        private final HadoopTaskOutput[] locAdders = new HadoopTaskOutput[locMaps.length()];

        /** */
        private final HadoopTaskOutput[] rmtAdders = new HadoopTaskOutput[rmtMaps.length()];

        /** Remote direct contexts. */
        private final HadoopDirectDataOutputContext[] rmtDirectCtxs =
            new HadoopDirectDataOutputContext[rmtMaps.length()];

        /** */
        private HadoopPartitioner partitioner;

        /** */
        private final HadoopTaskContext taskCtx;

        /**
         * Constructor.
         * @param taskCtx Task context.
         */
        private PartitionedOutput(HadoopTaskContext taskCtx) throws IgniteCheckedException {
            this.taskCtx = taskCtx;

            if (needPartitioner)
                partitioner = taskCtx.partitioner();
        }

        /** {@inheritDoc} */
        @Override public void write(Object key, Object val) throws IgniteCheckedException {
            int part = 0;

            if (partitioner != null) {
                part = partitioner.partition(key, val, totalReducerCnt);

                if (part < 0 || part >= totalReducerCnt)
                    throw new IgniteCheckedException("Invalid partition: " + part);
            }

            HadoopTaskOutput out;

            if (isLocalPartition(part)) {
                out = locAdders[part];

                if (out == null)
                    locAdders[part] = out = getOrCreateMap(locMaps, part).startAdding(taskCtx);
            }
            else {
                if (stripeMappers) {
                    int mapperIdx = HadoopMapperUtils.mapperIndex();

                    assert mapperIdx >= 0;

                    int idx = totalReducerCnt * mapperIdx + part;

                    HadoopDirectDataOutputContext rmtDirectCtx = rmtDirectCtxs[idx];

                    if (rmtDirectCtx == null) {
                        rmtDirectCtx = new HadoopDirectDataOutputContext(msgSize, msgGzip, taskCtx);

                        rmtDirectCtxs[idx] = rmtDirectCtx;
                    }

                    if (rmtDirectCtx.write(key, val))
                        sendShuffleMessage(idx, rmtDirectCtx, true);

                    return;
                }
                else {
                    out = rmtAdders[part];

                    if (out == null)
                        rmtAdders[part] = out = getOrCreateMap(rmtMaps, part).startAdding(taskCtx);
                }
            }

            out.write(key, val);
        }

        /** {@inheritDoc} */
        @Override public void onMapperFinished() throws IgniteCheckedException {
            if (stripeMappers) {
                int mapperIdx = HadoopMapperUtils.mapperIndex();

                assert mapperIdx >= 0;

                for (int i = 0; i < totalReducerCnt; i++) {
                    int idx = totalReducerCnt * mapperIdx + i;

                    sendShuffleMessage(idx, rmtDirectCtxs[idx], false);
                }
            }
        }

        /** {@inheritDoc} */
        @Override public void close() throws IgniteCheckedException {
            for (HadoopTaskOutput adder : locAdders) {
                if (adder != null)
                    adder.close();
            }

            for (HadoopTaskOutput adder : rmtAdders) {
                if (adder != null)
                    adder.close();
            }
        }
    }

    /**
     * Local task context proxy with delayed initialization.
     */
    private class LocalTaskContextProxy {
        /** Mutex for synchronization. */
        private final Object mux = new Object();

        /** Task info. */
        private final HadoopTaskInfo taskInfo;

        /** Task context. */
        private volatile HadoopTaskContext ctx;

        /**
         * Constructor.
         *
         * @param taskInfo Task info.
         */
        public LocalTaskContextProxy(HadoopTaskInfo taskInfo) {
            this.taskInfo = taskInfo;
        }

        /**
         * Get task context.
         *
         * @return Task context.
         * @throws IgniteCheckedException If failed.
         */
        public HadoopTaskContext get() throws IgniteCheckedException {
            HadoopTaskContext ctx0 = ctx;

            if (ctx0 == null) {
                synchronized (mux) {
                    ctx0 = ctx;

                    if (ctx0 == null) {
                        ctx0 = job.getTaskContext(taskInfo);

                        ctx = ctx0;
                    }
                }
            }

            return ctx0;
        }
    }
}


text:
package org.mozartoz.bootcompiler
package transform

import ast._

trait Walker {
  /** Transforms a Statement */
  def walkStat(statement: Statement): Unit = (statement: @unchecked) match {
    case CompoundStatement(stats) =>
      stats foreach walkStat

    case RawLocalStatement(declarations, body) =>
      declarations foreach walkDecl
      walkStat(body)

    case LocalStatement(declarations, body) =>
      walkStat(body)

    case CallStatement(callable, args) =>
      walkExpr(callable)
      args foreach walkExpr

    case IfStatement(condition, trueStatement, falseStatement) =>
      walkExpr(condition)
      walkStat(trueStatement)
      walkStat(falseStatement)

    case MatchStatement(value, clauses, elseStatement) =>
      walkExpr(value)
      clauses foreach walkClauseStat
      walkStat(elseStatement)

    case NoElseStatement() =>

    case ForStatement(from, to, proc) =>
      walkExpr(from)
      walkExpr(to)
      walkExpr(proc)

    case ThreadStatement(body) =>
      walkStat(body)

    case LockStatement(lock, body) =>
      walkExpr(lock)
      walkStat(body)

    case LockObjectStatement(body) =>
      walkStat(body)

    case TryStatement(body, exceptionVar, catchBody) =>
      walkStat(body)
      walkStat(catchBody)

    case TryFinallyStatement(body, finallyBody) =>
      walkStat(body)
      walkStat(finallyBody)

    case RaiseStatement(body) =>
      walkExpr(body)

    case FailStatement() =>

    case BindStatement(left, right) =>
      walkExpr(left)
      walkExpr(right)

    case BinaryOpStatement(left, operator, right) =>
      walkExpr(left)
      walkExpr(right)

    case DotAssignStatement(left, center, right) =>
      walkExpr(left)
      walkExpr(center)
      walkExpr(right)

    case SkipStatement() =>

    case ClearVarsStatement(stat, before, after) =>
      walkStat(stat)
  }

  def walkExpr(expression: Expression): Unit = (expression: @unchecked) match {
    case StatAndExpression(statement, expr) =>
      walkStat(statement)
      walkExpr(expr)

    case RawLocalExpression(declarations, expr) =>
      declarations foreach walkDecl
      walkExpr(expr)

    case LocalExpression(declarations, expr) =>
      walkExpr(expr)

    // Complex expressions

    case ProcExpression(name, args, body, flags) =>
      walkStat(body)

    case FunExpression(name, args, body, flags) =>
      walkExpr(body)

    case CallExpression(callable, args) =>
      walkExpr(callable)
      args foreach walkExpr

    case IfExpression(condition, trueExpression, falseExpression) =>
      walkExpr(condition)
      walkExpr(trueExpression)
      walkExpr(falseExpression)

    case MatchExpression(value, clauses, elseExpression) =>
      walkExpr(value)
      clauses foreach walkClauseExpr
      walkExpr(elseExpression)

    case NoElseExpression() =>

    case ThreadExpression(body) =>
      walkExpr(body)

    case LockExpression(lock, body) =>
      walkExpr(lock)
      walkExpr(body)

    case LockObjectExpression(body) =>
      walkExpr(body)

    case TryExpression(body, exceptionVar, catchBody) =>
      walkExpr(body)
      walkExpr(catchBody)

    case TryFinallyExpression(body, finallyBody) =>
      walkExpr(body)
      walkStat(finallyBody)

    case RaiseExpression(body) =>
      walkExpr(body)

    case BindExpression(left, right) =>
      walkExpr(left)
      walkExpr(right)

    case DotAssignExpression(left, center, right) =>
      walkExpr(left)
      walkExpr(center)
      walkExpr(right)

    case FunctorExpression(name, require, prepare, imports, define, exports) =>
      prepare foreach walkStat
      define foreach walkStat

    case ClearVarsExpression(expr, before, after) =>
      walkExpr(expr)

    // Operations

    case UnaryOp(operator, operand) =>
      walkExpr(operand)

    case BinaryOp(left, operator, right) =>
      walkExpr(left)
      walkExpr(right)

    case ShortCircuitBinaryOp(left, operator, right) =>
      walkExpr(left)
      walkExpr(right)

    // Trivial expressions

    case RawVariable(name) =>
    case Variable(symbol) =>
    case EscapedVariable(variable) =>
    case UnboundExpression() =>
    case NestingMarker() =>
    case Self() =>

    // Constants

    case Constant(value) =>

    // Records

    case AutoFeature() =>

    case Record(label, fields) =>
      walkExpr(label)
      fields foreach walkRecordField

    case OpenRecordPattern(label, fields) =>
      walkExpr(label)
      fields foreach walkRecordField

    case ListExpression(elements) =>
      elements foreach walkExpr

    case PatternConjunction(parts) =>
      parts foreach walkExpr

    // Classes

    case ClassExpression(name, parents, features, attributes, properties, methods) =>
      parents foreach walkExpr
      features foreach walkFeatOrAttr
      attributes foreach walkFeatOrAttr
      properties foreach walkExpr
      methods foreach walkMethodDef
  }

  /** Transforms a declaration */
  def walkDecl(declaration: RawDeclaration): Unit = declaration match {
    case stat: Statement => walkStat(stat)
    case _ =>
  }

  /** Transforms a record field */
  private def walkRecordField(field: RecordField): Unit = {
    walkExpr(field.feature)
    walkExpr(field.value)
  }

  /** Transforms a clause of a match statement */
  def walkClauseStat(clause: MatchStatementClause) = {
    walkExpr(clause.pattern)
    clause.guard foreach walkExpr
    walkStat(clause.body)
  }

  /** Transforms a clause of a match expression */
  def walkClauseExpr(clause: MatchExpressionClause) = {
    walkExpr(clause.pattern)
    clause.guard foreach walkExpr
    walkExpr(clause.body)
  }

  /** Transforms a feature or an attribute of a class */
  def walkFeatOrAttr(featOrAttr: FeatOrAttr) = {
    walkExpr(featOrAttr.name)
    featOrAttr.value foreach walkExpr
  }

  /** Transforms a method definition */
  def walkMethodDef(method: MethodDef): Unit = {
    method.body match {
      case stat: Statement => walkStat(stat)
      case expr: Expression => walkExpr(expr)
    }
  }
}

redpajama_set_name: RedPajamaGithub


text:
#pragma once
#include <aws/dynamodb/DynamoDB_EXPORTS.h>
#include <aws/dynamodb/model/TableDescription.h>
#include <utility>

namespace Aws
{
template<typename RESULT_TYPE>
class AmazonWebServiceResult;

namespace Utils
{
namespace Json
{
  class JsonValue;
} // namespace Json
} // namespace Utils
namespace DynamoDB
{
namespace Model
{
  class AWS_DYNAMODB_API RestoreTableFromBackupResult
  {
  public:
    RestoreTableFromBackupResult();
    RestoreTableFromBackupResult(const Aws::AmazonWebServiceResult<Aws::Utils::Json::JsonValue>& result);
    RestoreTableFromBackupResult& operator=(const Aws::AmazonWebServiceResult<Aws::Utils::Json::JsonValue>& result);

    /**
     *
The description of the table created from an existing backup.

     */
    inline const TableDescription& GetTableDescription() const{ return m_tableDescription; }

    /**
     *
The description of the table created from an existing backup.

     */
    inline void SetTableDescription(const TableDescription& value) { m_tableDescription = value; }

    /**
     *
The description of the table created from an existing backup.

     */
    inline void SetTableDescription(TableDescription&& value) { m_tableDescription = std::move(value); }

    /**
     *
The description of the table created from an existing backup.

     */
    inline RestoreTableFromBackupResult& WithTableDescription(const TableDescription& value) { SetTableDescription(value); return *this;}

    /**
     *
The description of the table created from an existing backup.

     */
    inline RestoreTableFromBackupResult& WithTableDescription(TableDescription&& value) { SetTableDescription(std::move(value)); return *this;}

  private:

    TableDescription m_tableDescription;
  };

} // namespace Model
} // namespace DynamoDB
} // namespace Aws


text:
import { Component, Input, Output } from '@angular/core';

@Component({
  selector: 'context-tab',
  styleUrls: ['./contextTab.component.css'],
  templateUrl: './contextTab.component.html'
})

export class ContextTabComponent {

  @Input() active: boolean;
  @Input() text: String;
  @Input() svgHeight: number;
  textY: number;
  transform: string;
  textColor: string;

  constructor() {
  }

  ngOnInit() {
    this.textY = this.svgHeight-15;
    this.transform = "rotate(-90 10, "+(this.svgHeight -20)+")";
  }

}


text:
namespace VolskSoft.ExTools
{
    using OfficeOpenXml.ConditionalFormatting.Contracts;

    public class ContainsRuleFormat : BaseRuleFormat, IExcelConditionalFormattingWithText
    {
        public ContainsRuleFormat(int columnNumber, string text, System.Drawing.Color? backgroundColor = null)
            :base(columnNumber, backgroundColor)
        {
            Text = text;
        }

        public string Text { get; set; }
    }
}


text:
var lineFinderSensor = require('jsupm_linefinder');

// Instantiate a  line finder sensor on digital pin D2
var myLineFinderSensor = new lineFinderSensor.LineFinder(2);

// Check every second for the presence of white detection
setInterval(function()
{
	if (myLineFinderSensor.whiteDetected())
		console.log("White detected.");
	else
		console.log("Black detected.");
}, 1000);

// Turn relay off when exiting
process.on('SIGINT', function()
{
	console.log("Exiting...");
	process.exit(0);
});


text:
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
// See the LICENSE file in the project root for more information.

/*============================================================
**
**
**
**
**
**
** Purpose: Exception for accessing a file that doesn't exist.
**
**
===========================================================*/

using System;
using System.Runtime.Serialization;
using SecurityException = System.Security.SecurityException;
using System.Globalization;

namespace System.IO
{
    // Thrown when trying to access a file that doesn't exist on disk.
    [Serializable]
    public class FileNotFoundException : IOException
    {
        private String _fileName;  // The name of the file that isn't found.
        private String _fusionLog;  // fusion log (when applicable)

        public FileNotFoundException()
            : base(Environment.GetResourceString("IO.FileNotFound"))
        {
            SetErrorCode(__HResults.COR_E_FILENOTFOUND);
        }

        public FileNotFoundException(String message)
            : base(message)
        {
            SetErrorCode(__HResults.COR_E_FILENOTFOUND);
        }

        public FileNotFoundException(String message, Exception innerException)
            : base(message, innerException)
        {
            SetErrorCode(__HResults.COR_E_FILENOTFOUND);
        }

        public FileNotFoundException(String message, String fileName) : base(message)
        {
            SetErrorCode(__HResults.COR_E_FILENOTFOUND);
            _fileName = fileName;
        }

        public FileNotFoundException(String message, String fileName, Exception innerException)
            : base(message, innerException)
        {
            SetErrorCode(__HResults.COR_E_FILENOTFOUND);
            _fileName = fileName;
        }

        public override String Message
        {
            get
            {
                SetMessageField();
                return _message;
            }
        }

        private void SetMessageField()
        {
            if (_message == null)
            {
                if ((_fileName == null) &&
                    (HResult == System.__HResults.COR_E_EXCEPTION))
                    _message = Environment.GetResourceString("IO.FileNotFound");

                else if (_fileName != null)
                    _message = FileLoadException.FormatFileLoadExceptionMessage(_fileName, HResult);
            }
        }

        public String FileName
        {
            get { return _fileName; }
        }

        public override String ToString()
        {
            String s = GetType().FullName + ": " + Message;

            if (_fileName != null && _fileName.Length != 0)
                s += Environment.NewLine + Environment.GetResourceString("IO.FileName_Name", _fileName);

            if (InnerException != null)
                s = s + " ---> " + InnerException.ToString();

            if (StackTrace != null)
                s += Environment.NewLine + StackTrace;

            try
            {
                if (FusionLog != null)
                {
                    if (s == null)
                        s = " ";
                    s += Environment.NewLine;
                    s += Environment.NewLine;
                    s += FusionLog;
                }
            }
            catch (SecurityException)
            {
            }
            return s;
        }

        protected FileNotFoundException(SerializationInfo info, StreamingContext context) : base(info, context)
        {
            // Base class constructor will check info != null.

            _fileName = info.GetString("FileNotFound_FileName");
            try
            {
                _fusionLog = info.GetString("FileNotFound_FusionLog");
            }
            catch
            {
                _fusionLog = null;
            }
        }

        private FileNotFoundException(String fileName, String fusionLog, int hResult)
            : base(null)
        {
            SetErrorCode(hResult);
            _fileName = fileName;
            _fusionLog = fusionLog;
            SetMessageField();
        }

        public String FusionLog
        {
            get { return _fusionLog; }
        }

        public override void GetObjectData(SerializationInfo info, StreamingContext context)
        {
            // Serialize data for our base classes.  base will verify info != null.
            base.GetObjectData(info, context);

            // Serialize data for this class
            info.AddValue("FileNotFound_FileName", _fileName, typeof(String));

            try
            {
                info.AddValue("FileNotFound_FusionLog", FusionLog, typeof(String));
            }
            catch (SecurityException)
            {
            }
        }
    }
}


text:
``` diff
+namespace System.Web.Script {
+    public class AjaxFrameworkAssemblyAttribute : Attribute {
+        public AjaxFrameworkAssemblyAttribute();
+        protected internal virtual Assembly GetDefaultAjaxFrameworkAssembly(Assembly currentAssembly);
+    }
+}
```


text:
(function(document) {
  'use strict';

  var app = document.querySelector('#app');
  app.firebaseURL = 'https://project-1728811666298556286.firebaseio.com';
  app.firebaseProvider = 'anonymous';

  app.items = [
    {
      done: true,
      text: 'Create Contacts'
    },
    {
      done: false,
      text: 'Use Firebase'
    }
  ];

  app.onFirebaseError = function(event) {
    this.$.errorToast.text = event.detail.message;
    this.$.errorToast.show();
  };
  app.onFirebaseLogin = function(event) {
    this.ref = new Firebase(this.firebaseURL + '/user/' +
                                                  event.detail.user.uid);
  };

  app.onFirebaseLogin = function(event) {
    this.ref = new Firebase(this.firebaseURL + '/user/' +
                                                    event.detail.user.uid);
    this.ref.on('value', function(snapshot) {
      app.updateItems(snapshot);
    });
  };

  app.addItem = function(event) {
    event.preventDefault();
    event.stopPropagation();
    this.ref.push({
      done: false,
      text: app.newItemValue
  });
    app.newItemValue = '';
  };

  app.toggleItem = function(event) {
    this.ref.
      child(event.model.item.uid).
      update({done: event.model.item.done});
  };

  app.deleteItem = function(event) {
    this.ref.child(event.model.item.uid).remove();
  };

})(document);


text:
package com.vdian.useful.domain;

import java.lang.reflect.Field;

/**
 * @author jifang
 * @since 16/7/22 下午5:57.
 */
public class Teacher extends User {

    private String name;

    public Teacher() {
    }

    public Teacher(String name) {
        this.name = name;
    }

    public static void main(String[] args) {
        Field[] fields = Teacher.class.getSuperclass().getDeclaredFields();
        for (Field field : fields) {
            System.out.println(field.getName());
        }
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }
}


text:
module TVF
  # class stores vocabulary information and associated fields
  class Vocabulary
    attr_reader :namespace, :uri, :fields
    def initialize(v)
      @namespace = v[:info][:namespace]
      @uri = v[:info][:uri]
      @fields = init_fields(v[:fields])
      add_field_methods
    end

    def valid?
      result = true
      result &&= fields_valid?
      result &&= namespace_valid?
      result && uri_valid?
    end

    def all
      fields.keys
    end

    def multiple
      fields.keys.select { |k| fields[k].multiple }
    end

    def single
      fields.keys.select { |k| !fields[k].multiple }
    end

    def facetable
      fields.keys.select { |k| fields[k].facetable }
    end

    private

    def init_fields(arg)
      hash = {}
      arg.each_pair do |k, v|
        hash[k] = Field.new(v)
      end
      hash
    end

    def add_field_methods
      fields.each_key do |k|
        define_singleton_method(k) do
          fields[k]
        end
      end
    end

    def fields_valid?
      fields.values.inject(true) { |a, e| a && e.valid? }
    end

    def namespace_valid?
      !namespace.nil?
    end

    def uri_valid?
      !uri.nil?
    end
  end
end


text:
package org.apache.stratos.rest.endpoint.api;

import org.apache.axis2.AxisFault;
import org.apache.axis2.context.ConfigurationContext;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.autoscaler.stub.AutoscalerServiceAutoScalingPolicyAlreadyExistExceptionException;
import org.apache.stratos.cloud.controller.stub.CloudControllerServiceCartridgeAlreadyExistsExceptionException;
import org.apache.stratos.cloud.controller.stub.CloudControllerServiceInvalidCartridgeDefinitionExceptionException;
import org.apache.stratos.cloud.controller.stub.CloudControllerServiceInvalidCartridgeTypeExceptionException;
import org.apache.stratos.cloud.controller.stub.CloudControllerServiceInvalidIaasProviderExceptionException;
import org.apache.stratos.cloud.controller.stub.domain.Cartridge;
import org.apache.stratos.common.beans.ResponseMessageBean;
import org.apache.stratos.common.beans.artifact.repository.GitNotificationPayloadBean;
import org.apache.stratos.common.beans.cartridge.CartridgeBean;
import org.apache.stratos.common.beans.partition.PartitionBean;
import org.apache.stratos.common.beans.policy.autoscale.AutoscalePolicyBean;
import org.apache.stratos.common.client.AutoscalerServiceClient;
import org.apache.stratos.common.client.CloudControllerServiceClient;
import org.apache.stratos.common.client.StratosManagerServiceClient;
import org.apache.stratos.manager.utils.ApplicationManagementUtil;
import org.apache.stratos.manager.utils.CartridgeConstants;
import org.apache.stratos.rest.endpoint.exception.RestAPIException;
import org.apache.stratos.rest.endpoint.util.converter.ObjectConverter;

import java.rmi.RemoteException;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Pattern;

/**
 * Utility class for API v4.0.
 * Please do not update this API utility class, if modifications are needed use the latest API version utility class.
 */
@Deprecated
public class StratosApiV40Utils {
    public static final String IS_VOLUME_REQUIRED = "volume.required";
    public static final String SHOULD_DELETE_VOLUME = "volume.delete.on.unsubscription";
    public static final String VOLUME_SIZE = "volume.size.gb";
    public static final String DEVICE_NAME = "volume.device.name";
    public static final String TENANT_RANGE_ALL = "*";

    private static Log log = LogFactory.getLog(StratosApiV40Utils.class);

    static ResponseMessageBean deployCartridge(CartridgeBean cartridgeDefinitionBean, ConfigurationContext ctxt,
                                               String userName, String tenantDomain) throws RestAPIException {

        log.info("Starting to deploy a Cartridge [type] " + cartridgeDefinitionBean.getType());

        CloudControllerServiceClient cloudControllerServiceClient = getCloudControllerServiceClient();

        if (cloudControllerServiceClient != null) {

            Cartridge cartridgeConfig = ObjectConverter.convertCartridgeBeanToStubCartridgeConfig
                    (cartridgeDefinitionBean, null, -1);

            if (cartridgeConfig == null) {
                throw new RestAPIException("Populated CartridgeConfig instance is null, cartridge deployment aborted");
            }

            // call CC
            try {
                cloudControllerServiceClient.addCartridge(cartridgeConfig);
            } catch (RemoteException e) {
                log.error(e.getMessage(), e);
                throw new RestAPIException(e.getMessage(), e);
            } catch (CloudControllerServiceInvalidCartridgeDefinitionExceptionException e) {
                String message = e.getFaultMessage().getInvalidCartridgeDefinitionException().getMessage();
                log.error(message, e);
                throw new RestAPIException(message, e);
            } catch (CloudControllerServiceCartridgeAlreadyExistsExceptionException e) {
                String message = e.getMessage();
                log.error(message, e);
                throw new RestAPIException(message, e);
            } catch (CloudControllerServiceInvalidIaasProviderExceptionException e) {
                String message = e.getFaultMessage().getInvalidIaasProviderException().getMessage();
                log.error(message, e);
                throw new RestAPIException(message, e);
            }
        }

        ResponseMessageBean stratosApiResponse = new ResponseMessageBean();
        String message = "Successfully added cartridge definition: [cartridge-type] " + cartridgeDefinitionBean.getType();
        stratosApiResponse.setMessage(message);
        if (log.isInfoEnabled()) {
            log.info(message);
        }
        return stratosApiResponse;
    }

//    @SuppressWarnings("unused")
//    private static DeploymentPolicy[] intersection(
//            DeploymentPolicy[] cartridgeDepPolicies,
//            DeploymentPolicy[] lbCartridgeDepPolicies) {
//
//        List<DeploymentPolicy> commonPolicies =
//                new ArrayList<DeploymentPolicy>();
//        for (DeploymentPolicy policy1
//                : cartridgeDepPolicies) {
//            for (DeploymentPolicy policy2
//                    : lbCartridgeDepPolicies) {
//                if(policy1.equals(policy2)) {
//                    commonPolicies.add(policy1);
//                }
//            }
//
//        }
//        return commonPolicies.toArray(new DeploymentPolicy[0]);
//    }

    static ResponseMessageBean undeployCartridge(String cartridgeType) throws RestAPIException {

        CloudControllerServiceClient cloudControllerServiceClient = getCloudControllerServiceClient();
        if (cloudControllerServiceClient != null) {
            try {
                cloudControllerServiceClient.removeCartridge(cartridgeType);
            } catch (RemoteException e) {
                log.error(e.getMessage(), e);
                throw new RestAPIException(e.getMessage(), e);
            } catch (CloudControllerServiceInvalidCartridgeTypeExceptionException e) {
                String msg = e.getFaultMessage().getInvalidCartridgeTypeException().getMessage();
                log.error(msg, e);
                throw new RestAPIException(msg, e);
            }

        }

        ResponseMessageBean stratosApiResponse = new ResponseMessageBean();
        stratosApiResponse.setMessage("Successfully undeployed cartridge definition with type " + cartridgeType);
        return stratosApiResponse;
    }

    public static ResponseMessageBean deployAutoscalingPolicy(AutoscalePolicyBean autoscalePolicyBean) throws RestAPIException {

        //log.info("***** " + cartridgeDefinitionBean.toString() + " *****");

        AutoscalerServiceClient autoscalerServiceClient = getAutoscalerServiceClient();
        if (autoscalerServiceClient != null) {

            org.apache.stratos.autoscaler.stub.autoscale.policy.AutoscalePolicy autoscalePolicy = ObjectConverter.
                    convertToCCAutoscalerPojo(autoscalePolicyBean,null,-1234);

            try {
                autoscalerServiceClient
                        .addAutoscalingPolicy(autoscalePolicy);
            } catch (RemoteException e) {
                log.error(e.getMessage(), e);
                throw new RestAPIException(e.getMessage(), e);
            } catch (AutoscalerServiceAutoScalingPolicyAlreadyExistExceptionException e) {
                log.error(e.getMessage(), e);
                throw new RestAPIException(e.getMessage(), e);
            }

        }

        ResponseMessageBean stratosApiResponse = new ResponseMessageBean();
        stratosApiResponse.setMessage("Successfully deployed autoscaling policy definition with id " + autoscalePolicyBean.getId());
        return stratosApiResponse;
    }

    private static CloudControllerServiceClient getCloudControllerServiceClient() throws RestAPIException {

        try {
            return CloudControllerServiceClient.getInstance();

        } catch (AxisFault axisFault) {
            String errorMsg = "Error while getting CloudControllerServiceClient instance to connect to the "
                    + "Cloud Controller. Cause: " + axisFault.getMessage();
            log.error(errorMsg, axisFault);
            throw new RestAPIException(errorMsg, axisFault);
        }
    }

    public static PartitionBean[] getAvailablePartitions() throws RestAPIException {

        org.apache.stratos.cloud.controller.stub.domain.Partition[] partitions = null;
        AutoscalerServiceClient autoscalerServiceClient = getAutoscalerServiceClient();
        if (autoscalerServiceClient != null) {
//            try {
//                partitions = autoscalerServiceClient.getAvailablePartitions();
//
//            } catch (RemoteException e) {
//                String errorMsg = "Error while getting available partitions. Cause : " + e.getMessage();
//                log.error(errorMsg, e);
//                throw new RestAPIException(errorMsg, e);
//            }
        }

        return ObjectConverter.populatePartitionPojos(partitions);
    }

    public static PartitionBean[] getPartitionsOfDeploymentPolicy(String deploymentPolicyId)
            throws RestAPIException {

        org.apache.stratos.cloud.controller.stub.domain.Partition[] partitions = null;
        AutoscalerServiceClient autoscalerServiceClient = getAutoscalerServiceClient();
//        if (autoscalerServiceClient != null) {
//            try {
//                partitions =
//                        autoscalerServiceClient.getPartitionsOfDeploymentPolicy(deploymentPolicyId);
//
//            } catch (RemoteException e) {
//                String errorMsg = "Error while getting available partitions for deployment policy id " +
//                        deploymentPolicyId+". Cause: "+e.getMessage();
//                log.error(errorMsg, e);
//                throw new RestAPIException(errorMsg, e);
//            }
//        }

        return ObjectConverter.populatePartitionPojos(partitions);
    }

    public static PartitionBean[] getPartitionsOfGroup(String deploymentPolicyId, String groupId) throws RestAPIException {

        org.apache.stratos.cloud.controller.stub.domain.Partition[] partitions = null;
        AutoscalerServiceClient autoscalerServiceClient = getAutoscalerServiceClient();
//        if (autoscalerServiceClient != null) {
//            try {
//                partitions =
//                        autoscalerServiceClient.getPartitionsOfGroup(deploymentPolicyId, groupId);
//
//            } catch (RemoteException e) {
//                String errorMsg = "Error while getting available partitions for deployment policy id " + deploymentPolicyId +
//                        ", group id " + groupId+". Cause: "+e.getMessage();
//                log.error(errorMsg, e);
//                throw new RestAPIException(errorMsg, e);
//            }
//        }

        return ObjectConverter.populatePartitionPojos(partitions);
    }

    public static PartitionBean getPartition(String partitionId) throws RestAPIException {

        org.apache.stratos.cloud.controller.stub.domain.Partition partition = null;
        AutoscalerServiceClient autoscalerServiceClient = getAutoscalerServiceClient();
//        if (autoscalerServiceClient != null) {
//            try {
//                partition = autoscalerServiceClient.getPartition(partitionId);
//
//            } catch (RemoteException e) {
//                String errorMsg = "Error while getting partition for id " + partitionId+". Cause: "+e.getMessage();
//                log.error(errorMsg, e);
//                throw new RestAPIException(errorMsg, e);
//            }
//        }

        return ObjectConverter.populatePartitionPojo(partition);
    }

    private static AutoscalerServiceClient getAutoscalerServiceClient() throws RestAPIException {

        try {
            return AutoscalerServiceClient.getInstance();
        } catch (AxisFault axisFault) {
            String errorMsg = "Error while getting AutoscalerServiceClient instance to connect to the "
                    + "Autoscaler";
            log.error(errorMsg, axisFault);
            throw new RestAPIException(errorMsg, axisFault);
        }
    }

    public static AutoscalePolicyBean[] getAutoScalePolicies() throws RestAPIException {

        org.apache.stratos.autoscaler.stub.autoscale.policy.AutoscalePolicy[] autoscalePolicies = null;
        AutoscalerServiceClient autoscalerServiceClient = getAutoscalerServiceClient();
        if (autoscalerServiceClient != null) {
            try {
                autoscalePolicies = autoscalerServiceClient.getAutoScalePolicies();

            } catch (RemoteException e) {
                String errorMsg = "Error while getting available autoscaling policies. Cause : " + e.getMessage();
                log.error(errorMsg, e);
                throw new RestAPIException(errorMsg, e);
            }
        }
        return ObjectConverter.convertStubAutoscalePoliciesToAutoscalePolicies(autoscalePolicies);
    }

    public static AutoscalePolicyBean getAutoScalePolicy(String autoscalePolicyId) throws RestAPIException {

        org.apache.stratos.autoscaler.stub.autoscale.policy.AutoscalePolicy autoscalePolicy = null;
        AutoscalerServiceClient autoscalerServiceClient = getAutoscalerServiceClient();
        if (autoscalerServiceClient != null) {
            try {
                autoscalePolicy = autoscalerServiceClient.getAutoScalePolicy(autoscalePolicyId);

            } catch (RemoteException e) {
                String errorMsg = "Error while getting information for autoscaling policy with id " +
                        autoscalePolicyId + ".  Cause: " + e.getMessage();
                log.error(errorMsg, e);
                throw new RestAPIException(errorMsg, e);
            }
        }

        return ObjectConverter.convertStubAutoscalePolicyToAutoscalePolicy(autoscalePolicy);
    }

//    public static DeploymentPolicyBean getDeploymentPolicy(String deploymentPolicyId) throws RestAPIException {
//
//        DeploymentPolicy deploymentPolicy = null;
//
//
//        return null;
//    }

    static CartridgeBean getAvailableCartridgeInfo(String cartridgeType, Boolean multiTenant, ConfigurationContext configurationContext) throws RestAPIException {
        List<CartridgeBean> cartridges = getAvailableCartridges(null, multiTenant, configurationContext);
        for (CartridgeBean cartridge : cartridges) {
            if (cartridge.getType().equals(cartridgeType)) {
                return cartridge;
            }
        }
        String msg = "Unavailable cartridge type: " + cartridgeType;
        log.error(msg);
        throw new RestAPIException(msg);
    }

    static List<CartridgeBean> getAvailableLbCartridges(Boolean multiTenant,
                                                        ConfigurationContext configurationContext) throws RestAPIException {
        List<CartridgeBean> cartridges = getAvailableCartridges(null, multiTenant,
                configurationContext);
        List<CartridgeBean> lbCartridges = new ArrayList<CartridgeBean>();
        for (CartridgeBean cartridge : cartridges) {
            if (cartridge.getCategory().equals("LoadBalancer")) {
                lbCartridges.add(cartridge);
            }
        }

		/*if(lbCartridges == null || lbCartridges.isEmpty()) {
            String msg = "Load balancer Cartridges are not available.";
	        log.error(msg);
	        throw new RestAPIException(msg) ;
		}*/
        return lbCartridges;
    }

    static List<CartridgeBean> getAvailableCartridges(String cartridgeSearchString, Boolean multiTenant, ConfigurationContext configurationContext) throws RestAPIException {
        List<CartridgeBean> cartridges = new ArrayList<CartridgeBean>();

        if (log.isDebugEnabled()) {
            log.debug("Getting available cartridges. Search String: " + cartridgeSearchString + ", Multi-Tenant: " + multiTenant);
        }

        boolean allowMultipleSubscription = Boolean.valueOf(System.getProperty(CartridgeConstants.FEATURE_MULTI_TENANT_MULTIPLE_SUBSCRIPTION_ENABLED));

        try {
            Pattern searchPattern = getSearchStringPattern(cartridgeSearchString);

            String[] availableCartridges = CloudControllerServiceClient.getInstance().getRegisteredCartridges();

            if (availableCartridges != null) {
                for (String cartridgeType : availableCartridges) {
                    Cartridge cartridgeInfo = null;
                    try {
                        cartridgeInfo = CloudControllerServiceClient.getInstance().getCartridgeByTenant(cartridgeType,-1234);
                    } catch (Exception e) {
                        if (log.isWarnEnabled()) {
                            log.warn("Error when calling getCartridgeInfo for " + cartridgeType + ", Error: "
                                    + e.getMessage());
                        }
                    }
                    if (cartridgeInfo == null) {
                        // This cannot happen. But continue
                        if (log.isDebugEnabled()) {
                            log.debug("Cartridge Info not found: " + cartridgeType);
                        }
                        continue;
                    }

                    if (multiTenant != null && !multiTenant && cartridgeInfo.getMultiTenant()) {
                        // Need only Single-Tenant cartridges
                        continue;
                    } else if (multiTenant != null && multiTenant && !cartridgeInfo.getMultiTenant()) {
                        // Need only Multi-Tenant cartridges
                        continue;
                    }

                    if (!StratosApiV40Utils.cartridgeMatches(cartridgeInfo, searchPattern)) {
                        continue;
                    }

                    CartridgeBean cartridge = new CartridgeBean();
                    cartridge.setType(cartridgeType);
                    cartridge.setProvider(cartridgeInfo.getProvider());
                    cartridge.setDisplayName(cartridgeInfo.getDisplayName());
                    cartridge.setDescription(cartridgeInfo.getDescription());
                    cartridge.setVersion(cartridgeInfo.getVersion());
                    cartridge.setMultiTenant(cartridgeInfo.getMultiTenant());
                    cartridge.setHost(cartridgeInfo.getHostName());
                    cartridges.add(cartridge);

                    if (cartridgeInfo.getMultiTenant() && !allowMultipleSubscription) {
                        // If the cartridge is multi-tenant. We should not let users
                        // createSubscription twice.
                        if (isAlreadySubscribed(cartridgeType,
                                ApplicationManagementUtil.getTenantId(configurationContext))) {
                            if (log.isDebugEnabled()) {
                                log.debug("Already subscribed to " + cartridgeType
                                        + ". This multi-tenant cartridge will not be available to createSubscription");
                            }
                        }
                    }
                }
            } else {
                if (log.isDebugEnabled()) {
                    log.debug("There are no available cartridges");
                }
            }
        } catch (Exception e) {
            String msg = "Error while getting available cartridges. Cause: " + e.getMessage();
            log.error(msg, e);
            throw new RestAPIException(msg, e);
        }

        //Collections.sort(cartridges);

        if (log.isDebugEnabled()) {
            log.debug("Returning available cartridges " + cartridges.size());
        }

        return cartridges;
    }

    private static boolean isAlreadySubscribed(String cartridgeType, int tenantId) {

//        Collection<CartridgeSubscription> subscriptionList = cartridgeSubsciptionManager.isCartridgeSubscribed(tenantId, cartridgeType);
//        if(subscriptionList == null || subscriptionList.isEmpty()){
//            return false;
//        }else {
//            return true;
//        }
        return false;
    }

//    public static List<ServiceDefinitionBean> getdeployedServiceInformation () throws RestAPIException {
//
//        Collection<Service> services = null;
//
//        try {
//            services = serviceDeploymentManager.getServices();
//
//        } catch (ADCException e) {
//            String msg = "Unable to get deployed service information. Cause: "+e.getMessage();
//            log.error(msg, e);
//            throw new RestAPIException(msg, e);
//        }
//
//        if (services != null && !services.isEmpty()) {
//            return ObjectConverter.convertToServiceDefinitionBeans(services);
//        }
//
//        return null;
//    }

//    public static ServiceDefinitionBean getDeployedServiceInformation (String type) throws RestAPIException {
//
//        Service service = null;
//
//        try {
//            service = serviceDeploymentManager.getService(type);
//
//        } catch (ADCException e) {
//            String msg = "Unable to get deployed service information for [type]: " + type+". Cause: "+e.getMessage();
//            log.error(msg, e);
//            throw new RestAPIException(msg, e);
//        }
//
//        if (service != null) {
//            return ObjectConverter.convertToServiceDefinitionBean(service);
//        }
//
//        return new ServiceDefinitionBean();
//    }

//    public static List<Cartridge> getActiveDeployedServiceInformation (ConfigurationContext configurationContext) throws RestAPIException {
//
//        Collection<Service> services = null;
//
//        try {
//            services = serviceDeploymentManager.getServices();
//
//        } catch (ADCException e) {
//            String msg = "Unable to get deployed service information. Cause: "+e.getMessage();
//            log.error(msg, e);
//            throw new RestAPIException(msg, e);
//        }
//
//        List<Cartridge> availableMultitenantCartridges = new ArrayList<Cartridge>();
//        int tenantId = ApplicationManagementUtil.getTenantId(configurationContext);
//        //getting the services for the tenantId
//        for(Service service : services) {
//            String tenantRange = service.getTenantRange();
//            if(tenantRange.equals(TENANT_RANGE_ALL)) {
//                //check whether any active instances found for this service in the Topology
//
//                Cluster cluster = TopologyManager.getTopology().getService(service.getType()).
//                        getCluster(service.getClusterId());
//                boolean activeMemberFound = false;
//                for(Member member : cluster.getMembers()) {
//                    if(member.isActive()) {
//                        activeMemberFound = true;
//                        break;
//                    }
//                }
//                if(activeMemberFound) {
//                    availableMultitenantCartridges.add(getAvailableCartridgeInfo(null, true, configurationContext));
//                }
//            } else {
//                //TODO have to check for the serivces which has correct tenant range
//            }
//        }
//
//		/*if (availableMultitenantCartridges.isEmpty()) {
//			String msg = "Cannot find any active deployed service for tenant [id] "+tenantId;
//			log.error(msg);
//			throw new RestAPIException(msg);
//		}*/
//
//        return availableMultitenantCartridges;
//    }

//    static List<Cartridge> getSubscriptions (String cartridgeSearchString, String serviceGroup, ConfigurationContext configurationContext) throws RestAPIException {
//
//        List<Cartridge> cartridges = new ArrayList<Cartridge>();
//
//        if (log.isDebugEnabled()) {
//            log.debug("Getting subscribed cartridges. Search String: " + cartridgeSearchString);
//        }
//
//        try {
//            Pattern searchPattern = getSearchStringPattern(cartridgeSearchString);
//
//            Collection<CartridgeSubscription> subscriptions = cartridgeSubsciptionManager.getCartridgeSubscriptions(ApplicationManagementUtil.
//                    getTenantId(configurationContext), null);
//
//            if (subscriptions != null && !subscriptions.isEmpty()) {
//
//                for (CartridgeSubscription subscription : subscriptions) {
//
//                    if (!cartridgeMatches(subscription.getCartridgeInfo(), subscription, searchPattern)) {
//                        continue;
//                    }
//                    Cartridge cartridge = getCartridgeFromSubscription(subscription);
//                    if (cartridge == null) {
//                        continue;
//                    }
//                    Cluster cluster = TopologyClusterInformationModel.getInstance().getCluster(ApplicationManagementUtil.getTenantId(configurationContext),
//                            cartridge.getCartridgeAlias());
//                    String cartridgeStatus = "Inactive";
//                    int activeMemberCount = 0;
//                    if (cluster != null) {
//                        Collection<Member> members = cluster.getMembers();
//                        for (Member member : members) {
//                            if (member.isActive()) {
//                                cartridgeStatus = "Active";
//                                activeMemberCount++;
//                            }
//                        }
//                    }
//                    cartridge.setActiveInstances(activeMemberCount);
//                    cartridge.setStatus(cartridgeStatus);
//
//                    // Ignoring the LB cartridges since they are not shown to the user.
//                    if(cartridge.isLoadBalancer())
//                        continue;
//                    if(StringUtils.isNotEmpty(serviceGroup) && cartridge.getServiceGroup() != null &&
//                            !cartridge.getServiceGroup().equals(serviceGroup)){
//                        continue;
//                    }
//                    cartridges.add(cartridge);
//                }
//            } else {
//                if (log.isDebugEnabled()) {
//                    log.debug("There are no subscribed cartridges");
//                }
//            }
//        } catch (Exception e) {
//            String msg = "Error while getting subscribed cartridges. Cause: "+e.getMessage();
//            log.error(msg, e);
//            throw new RestAPIException(msg, e);
//        }
//
//        Collections.sort(cartridges);
//
//        if (log.isDebugEnabled()) {
//            log.debug("Returning subscribed cartridges " + cartridges.size());
//        }
//
//        /*if(cartridges.isEmpty()) {
//        	String msg = "Cannot find any subscribed Cartridge, matching the given string: "+cartridgeSearchString;
//            log.error(msg);
//            throw new RestAPIException(msg);
//        }*/
//
//        return cartridges;
//    }

//    static Cartridge getSubscription(String cartridgeAlias, ConfigurationContext configurationContext) throws RestAPIException {
//
//        Cartridge cartridge =  getCartridgeFromSubscription(cartridgeSubsciptionManager.getCartridgeSubscription(ApplicationManagementUtil.
//                getTenantId(configurationContext), cartridgeAlias));
//
//        if (cartridge == null) {
//            String message = "Unregistered [alias]: "+cartridgeAlias+"! Please enter a valid alias.";
//            log.error(message);
//            throw new RestAPIException(Response.Status.NOT_FOUND, message);
//        }
//        Cluster cluster = TopologyClusterInformationModel.getInstance().getCluster(ApplicationManagementUtil.getTenantId(configurationContext),
//                cartridge.getCartridgeAlias());
//        String cartridgeStatus = "Inactive";
//        int activeMemberCount = 0;
//
//        // cluster might not be created yet, so need to check
//        if (cluster != null) {
//            Collection<Member> members = cluster.getMembers();
//            if (members != null ) {
//                for (Member member : members) {
//                    if(member.isActive()) {
//                        cartridgeStatus = "Active";
//                        activeMemberCount++;
//                    }
//                }
//            }
//        }
//
//        cartridge.setActiveInstances(activeMemberCount);
//        cartridge.setStatus(cartridgeStatus);
//        return cartridge;
//
//    }

//    static int getActiveInstances(String cartridgeType, String cartridgeAlias, ConfigurationContext configurationContext) throws RestAPIException {
//        int noOfActiveInstances = 0;
//        Cluster cluster = TopologyClusterInformationModel.getInstance().getCluster(ApplicationManagementUtil.getTenantId(configurationContext),
//                cartridgeAlias);
//
//        if(cluster == null) {
//            String message = "No Cluster found for cartridge [type] "+cartridgeType+", [alias] "+cartridgeAlias;
//            log.error(message);
//            throw new RestAPIException(message);
//        }
//
//        for(Member member : cluster.getMembers()) {
//            if(member.getStatus().toString().equals(MemberStatus.Active)) {
//                noOfActiveInstances ++;
//            }
//        }
//        return noOfActiveInstances;
//    }

//    private static Cartridge getCartridgeFromSubscription(CartridgeSubscription subscription) throws RestAPIException {
//
//        if (subscription == null) {
//            return null;
//        }
//        try {
//            Cartridge cartridge = new Cartridge();
//            cartridge.setCartridgeType(subscription.getCartridgeInfo()
//                    .getType());
//            cartridge.setMultiTenant(subscription.getCartridgeInfo()
//                    .getMultiTenant());
//            cartridge
//                    .setProvider(subscription.getCartridgeInfo().getProvider());
//            cartridge.setVersion(subscription.getCartridgeInfo().getVersion());
//            cartridge.setDescription(subscription.getCartridgeInfo()
//                    .getDescription());
//            cartridge.setDisplayName(subscription.getCartridgeInfo()
//                    .getDisplayName());
//            cartridge.setCartridgeAlias(subscription.getAlias());
//            cartridge.setHostName(subscription.getHostName());
//            cartridge.setMappedDomain(subscription.getMappedDomain());
//            if (subscription.getRepository() != null) {
//                cartridge.setRepoURL(subscription.getRepository().getUrl());
//            }
//
//            if (subscription instanceof DataCartridgeSubscription) {
//                DataCartridgeSubscription dataCartridgeSubscription = (DataCartridgeSubscription) subscription;
//                cartridge.setDbHost(dataCartridgeSubscription.getDBHost());
//                cartridge.setDbUserName(dataCartridgeSubscription
//                        .getDBUsername());
//                cartridge
//                        .setPassword(dataCartridgeSubscription.getDBPassword());
//            }
//
//            if (subscription.getLbClusterId() != null
//                    && !subscription.getLbClusterId().isEmpty()) {
//                cartridge.setLbClusterId(subscription.getLbClusterId());
//            }
//
//            cartridge.setStatus(subscription.getSubscriptionStatus());
//            cartridge.setPortMappings(subscription.getCartridgeInfo()
//                    .getPortMappings());
//
//            if(subscription.getCartridgeInfo().getLbConfig() != null && subscription.getCartridgeInfo().getProperties() != null) {
//                for(org.apache.stratos.cloud.controller.stub.Property property: subscription.getCartridgeInfo().getProperties()) {
//                    if(property.getName().equals("load.balancer")) {
//                        cartridge.setLoadBalancer(true);
//                    }
//                }
//            }
//            if(subscription.getCartridgeInfo().getServiceGroup() != null) {
//                cartridge.setServiceGroup(subscription.getCartridgeInfo().getServiceGroup());
//            }
//            return cartridge;
//
//        } catch (Exception e) {
//            String msg = "Unable to extract the Cartridge from subscription. Cause: "+e.getMessage();
//            log.error(msg);
//            throw new RestAPIException(msg);
//        }
//
//    }

    static Pattern getSearchStringPattern(String searchString) {
        if (log.isDebugEnabled()) {
            log.debug("Creating search pattern for " + searchString);
        }
        if (searchString != null) {
            // Copied from org.wso2.carbon.webapp.mgt.WebappAdmin.doesWebappSatisfySearchString(WebApplication, String)
            String regex = searchString.toLowerCase().replace("..?", ".?").replace("..*", ".*").replaceAll("\\?", ".?")
                    .replaceAll("\\*", ".*?");
            if (log.isDebugEnabled()) {
                log.debug("Created regex: " + regex + " for search string " + searchString);
            }

            Pattern pattern = Pattern.compile(regex);
            return pattern;
        }
        return null;
    }

    static boolean cartridgeMatches(Cartridge cartridgeInfo, Pattern pattern) {
        if (pattern != null) {
            boolean matches = false;
            if (cartridgeInfo.getDisplayName() != null) {
                matches = pattern.matcher(cartridgeInfo.getDisplayName().toLowerCase()).find();
            }
            if (!matches && cartridgeInfo.getDescription() != null) {
                matches = pattern.matcher(cartridgeInfo.getDescription().toLowerCase()).find();
            }
            return matches;
        }
        return true;
    }

//    static boolean cartridgeMatches(CartridgeInfo cartridgeInfo, CartridgeSubscription cartridgeSubscription, Pattern pattern) {
//        if (pattern != null) {
//            boolean matches = false;
//            if (cartridgeInfo.getDisplayName() != null) {
//                matches = pattern.matcher(cartridgeInfo.getDisplayName().toLowerCase()).find();
//            }
//            if (!matches && cartridgeInfo.getDescription() != null) {
//                matches = pattern.matcher(cartridgeInfo.getDescription().toLowerCase()).find();
//            }
//            if (!matches && cartridgeSubscription.getType() != null) {
//                matches = pattern.matcher(cartridgeSubscription.getType().toLowerCase()).find();
//            }
//            if (!matches && cartridgeSubscription.getAlias() != null) {
//                matches = pattern.matcher(cartridgeSubscription.getAlias().toLowerCase()).find();
//            }
//            return matches;
//        }
//        return true;
//    }

//    public static CartridgeSubscription getCartridgeSubscription(String alias, ConfigurationContext configurationContext) {
//        return cartridgeSubsciptionManager.getCartridgeSubscription(ApplicationManagementUtil.getTenantId(configurationContext), alias);
//    }
//
//    public static org.apache.stratos.common.beans.topology.Cluster getCluster (String cartridgeType, String subscriptionAlias, ConfigurationContext configurationContext) throws RestAPIException {
//
//        Cluster cluster = TopologyClusterInformationModel.getInstance().getCluster(ApplicationManagementUtil.getTenantId(configurationContext),
//                subscriptionAlias);
//        if(cluster == null) {
//            throw new RestAPIException("No matching cluster found for [cartridge type]: "+cartridgeType+ " [alias] "+subscriptionAlias);
//        } else{
//            return ObjectConverter.convertClusterToClusterBean(cluster, null);
//        }
//    }

//    public static org.apache.stratos.common.beans.topology.Cluster[] getClustersForTenant (ConfigurationContext configurationContext) {
//
//        Set<Cluster> clusterSet = TopologyClusterInformationModel.getInstance().getClusters(ApplicationManagementUtil.
//                getTenantId(configurationContext), null);
//        ArrayList<org.apache.stratos.common.beans.topology.Cluster> clusters =
//                new ArrayList<org.apache.stratos.common.beans.topology.Cluster>();
//        for(Cluster cluster : clusterSet) {
//            clusters.add(ObjectConverter.convertClusterToClusterBean(cluster, null));
//        }
//        org.apache.stratos.common.beans.topology.Cluster[] arrCluster =
//                new org.apache.stratos.common.beans.topology.Cluster[clusters.size()];
//        arrCluster = clusters.toArray(arrCluster);
//        return arrCluster;
//
//    }

//    public static org.apache.stratos.common.beans.topology.Cluster[] getClustersForTenantAndCartridgeType (ConfigurationContext configurationContext,
//                                                                                String cartridgeType) {
//
//        Set<Cluster> clusterSet = TopologyClusterInformationModel.getInstance().getClusters(ApplicationManagementUtil.
//                getTenantId(configurationContext), cartridgeType);
//        List<org.apache.stratos.common.beans.topology.Cluster> clusters =
//                new ArrayList<org.apache.stratos.common.beans.topology.Cluster>();
//        for(Cluster cluster : clusterSet) {
//            clusters.add(ObjectConverter.convertClusterToClusterBean(cluster, null));
//        }
//        org.apache.stratos.common.beans.topology.Cluster[] arrCluster =
//                new org.apache.stratos.common.beans.topology.Cluster[clusters.size()];
//        arrCluster = clusters.toArray(arrCluster);
//        return arrCluster;
//
//    }

//    public static org.apache.stratos.common.beans.topology.Cluster[] getClustersForCartridgeType(String cartridgeType) {
//
//        Set<Cluster> clusterSet = TopologyClusterInformationModel
//                .getInstance()
//                .getClusters(cartridgeType);
//        List<org.apache.stratos.common.beans.topology.Cluster> clusters = new ArrayList<org.apache.stratos.common.beans.topology.Cluster>();
//        for (Cluster cluster : clusterSet) {
//            clusters.add(ObjectConverter.convertClusterToClusterBean(cluster, null));
//        }
//        org.apache.stratos.common.beans.topology.Cluster[] arrCluster = new org.apache.stratos.common.beans.topology.Cluster[clusters
//                .size()];
//        arrCluster = clusters.toArray(arrCluster);
//        return arrCluster;
//
//    }

    // return the cluster id for the lb. This is a temp fix.
    /*private static String subscribeToLb(String cartridgeType, String loadBalancedCartridgeType, String lbAlias,
        String defaultAutoscalingPolicy, String deploymentPolicy,
        ConfigurationContext configurationContext, String userName, String tenantDomain, Property[] props) throws ADCException {
        CartridgeSubscription cartridgeSubscription;
        try {
            if(log.isDebugEnabled()) {
                log.debug("Subscribing to a load balancer [cartridge] "+cartridgeType+" [alias] "+lbAlias);
            }
            SubscriptionData subscriptionData = new SubscriptionData();
            subscriptionData.setCartridgeType(cartridgeType);
            subscriptionData.setCartridgeAlias(lbAlias.trim());
            subscriptionData.setAutoscalingPolicyName(defaultAutoscalingPolicy);
            subscriptionData.setDeploymentPolicyName(deploymentPolicy);
            subscriptionData.setTenantDomain(tenantDomain);
            subscriptionData.setTenantId(ApplicationManagementUtil.getTenantId(configurationContext));
            subscriptionData.setTenantAdminUsername(userName);
            subscriptionData.setRepositoryType("git");
            //subscriptionData.setProperties(props);
            subscriptionData.setPrivateRepository(false);
            cartridgeSubscription =
                    cartridgeSubsciptionManager.subscribeToCartridgeWithProperties(subscriptionData);
            //set a payload parameter to indicate the load balanced cartridge type
            cartridgeSubscription.getPayloadData().add("LOAD_BALANCED_SERVICE_TYPE", loadBalancedCartridgeType);
            Properties lbProperties = new Properties();
            lbProperties.setProperties(props);
            cartridgeSubsciptionManager.registerCartridgeSubscription(cartridgeSubscription, lbProperties);

            if(log.isDebugEnabled()) {
                log.debug("Successfully subscribed to a load balancer [cartridge] "+cartridgeType+" [alias] "+lbAlias);
            }
        } catch (Exception e) {
            String msg = "Error while subscribing to load balancer cartridge [type] "+cartridgeType+". Cause: "+e.getMessage();
            log.error(msg, e);
            throw new ADCException(msg, e);
        }
        return cartridgeSubscription.getClusterDomain();
    } */

    static ResponseMessageBean unsubscribe(String alias, String tenantDomain) throws RestAPIException {
        throw new RestAPIException("Not implemented");
    }

    static void notifyArtifactUpdatedEvent(GitNotificationPayloadBean payload) throws RestAPIException {
        try {
            StratosManagerServiceClient serviceClient = StratosManagerServiceClient.getInstance();
            serviceClient.notifyArtifactUpdatedEventForRepository(payload.getRepository().getUrl());
        } catch (Exception e) {
            String message = "Could not send artifact updated event";
            log.error(message, e);
            throw new RestAPIException(message, e);
        }
    }

    public static ResponseMessageBean removeSubscriptionDomain(ConfigurationContext configurationContext, String cartridgeType,
                                                               String subscriptionAlias, String domain) throws RestAPIException {
        throw new RestAPIException("Not implemented");
    }
}


text:
{% extends "layout_citizen.html" %} {% block page_title %} Search for a charge {% endblock %} {% block content %}

            BETA
            This is a new service – your feedback will help us to improve it.

                Search for local land charges (LLC) on land or property

Use this service to find legal rights and restrictions (local land charges) on a property.

Search for free to see local land charges on any property you want to buy

Pay £15 to get an official search

                Start now

Before you start

About local land charges

Local land charges (LLC) are legal rights and restrictions on land. For example, a tree preservation order is a local land charge that stops you from cutting down a tree.

Why you need to search for local land charges

If you are thinking about buying a house, you need to find out if it has any local land charges on it. This is because local land charges affect what you can do on a property.

This search is usually done by a conveyancer, but HM Land Registry also offers a search of the local land charges register for free. Use the free search to find out more about the property, to help you decide if you want to buy it.

Difference between the free and paid searches

                        Free service
                        £15 service

where the local land charge is

description of the local land charge (what it does)

registration date (when it starts)

contact details for a local authority (they can answer questions about this local land charge)

where the local land charge is

description of the local land charge (what it does)

registration date (when it starts)

contact details for a local authority (they can answer questions about this local land charge)

official search, guaranteed by HM Land Registry

 link to documents (you need this if you want to research a charge)

local land charge boundary (shows where a charge starts and stops)

for 6 months:

refresh your search (make sure no new local land charges have been added before you exchange contracts)

return to your result (in case you lose your search result)

download your results in different formats (PDF, JSON, CSV, XML)

{% endblock %}