text:
/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.facebook.presto.execution.buffer;

import com.facebook.presto.OutputBuffers;
import com.facebook.presto.OutputBuffers.OutputBufferId;
import com.facebook.presto.block.BlockAssertions;
import com.facebook.presto.execution.TaskId;
import com.facebook.presto.operator.PageAssertions;
import com.facebook.presto.spi.Page;
import com.facebook.presto.spi.type.BigintType;
import com.facebook.presto.spi.type.Type;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.ListenableFuture;
import io.airlift.units.DataSize;
import io.airlift.units.Duration;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import static com.facebook.presto.OutputBuffers.BufferType.BROADCAST;
import static com.facebook.presto.OutputBuffers.BufferType.PARTITIONED;
import static com.facebook.presto.OutputBuffers.createInitialEmptyOutputBuffers;
import static com.facebook.presto.execution.buffer.BufferResult.emptyResults;
import static com.facebook.presto.spi.type.BigintType.BIGINT;
import static com.google.common.base.Preconditions.checkArgument;
import static io.airlift.concurrent.MoreFutures.tryGetFutureValue;
import static io.airlift.concurrent.Threads.daemonThreadsNamed;
import static io.airlift.units.DataSize.Unit.BYTE;
import static java.util.concurrent.Executors.newScheduledThreadPool;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;

public class TestSharedOutputBuffer
{
    private static final Duration NO_WAIT = new Duration(0, TimeUnit.MILLISECONDS);
    private static final Duration MAX_WAIT = new Duration(1, TimeUnit.SECONDS);
    private static final DataSize PAGE_SIZE = new DataSize(createPage(42).getSizeInBytes(), BYTE);
    private static final DataSize RETAINED_PAGE_SIZE = new DataSize(createPage(42).getRetainedSizeInBytes(), BYTE);
    private static final TaskId TASK_ID = new TaskId("query", 0, 0);
    private static final int DEFAULT_PARTITION = 0;
    private static final String TASK_INSTANCE_ID = "task-instance-id";

    private static final ImmutableList<BigintType> TYPES = ImmutableList.of(BIGINT);
    public static final OutputBufferId FIRST = new OutputBufferId(0);
    public static final OutputBufferId SECOND = new OutputBufferId(1);
    public static final OutputBufferId QUEUE = new OutputBufferId(0);
    public static final OutputBufferId FOO = new OutputBufferId(1);

    private static Page createPage(int i)
    {
        return new Page(BlockAssertions.createLongsBlock(i));
    }

    public static DataSize sizeOfPages(int count)
    {
        return new DataSize(PAGE_SIZE.toBytes() * count, BYTE);
    }

    public static DataSize retainedSizeOfPages(int count)
    {
        return new DataSize(RETAINED_PAGE_SIZE.toBytes() * count, BYTE);
    }

    private ScheduledExecutorService stateNotificationExecutor;

    @BeforeClass
    public void setUp()
            throws Exception
    {
        stateNotificationExecutor = newScheduledThreadPool(5, daemonThreadsNamed("test-%s"));
    }

    @AfterClass
    public void tearDown()
            throws Exception
    {
        if (stateNotificationExecutor != null) {
            stateNotificationExecutor.shutdownNow();
            stateNotificationExecutor = null;
        }
    }

    @Test
    public void testInvalidConstructorArg()
            throws Exception
    {
        try {
            new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, new DataSize(0, BYTE));
            fail("Expected IllegalStateException");
        }
        catch (IllegalArgumentException ignored) {
        }
    }

    @Test
    public void testSimple()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));

        // add three items
        for (int i = 0; i < 3; i++) {
            addPage(sharedBuffer, createPage(i));
        }

        OutputBuffers outputBuffers = createInitialEmptyOutputBuffers(BROADCAST).withBuffer(FIRST, 0);

        // add a queue
        sharedBuffer.setOutputBuffers(outputBuffers);
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 3, 0, 3, 3);

        // get the three elements
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 0, sizeOfPages(10), NO_WAIT), bufferResult(0, createPage(0), createPage(1), createPage(2)));
        // pages not acknowledged yet so state is the same
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 3, 0, 3, 3);

        // acknowledge first three pages
        sharedBuffer.get(FIRST, 3, sizeOfPages(10)).cancel(true);
        // pages now acknowledged
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 0, 3, 3, 3);

        // fill the buffer (we already added 3 pages)
        for (int i = 3; i < 10; i++) {
            addPage(sharedBuffer, createPage(i));
        }
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 7, 3, 10, 10);

        // try to add one more page, which should block
        ListenableFuture<?> future = enqueuePage(sharedBuffer, createPage(10));
        assertFalse(future.isDone());
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 8, 3, 11, 11);

        // remove a page
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 3, sizeOfPages(1), NO_WAIT), bufferResult(3, createPage(3)));
        // page not acknowledged yet so sent count is the same
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 8, 3, 11, 11);

        // we should still be blocked
        assertFalse(future.isDone());

        //
        // add another buffer and verify it sees all pages
        outputBuffers = outputBuffers.withBuffer(SECOND, 0);
        sharedBuffer.setOutputBuffers(outputBuffers);
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 11, 0, 11, 11);
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, SECOND, 0, sizeOfPages(10), NO_WAIT), bufferResult(0, createPage(0),
                createPage(1),
                createPage(2),
                createPage(3),
                createPage(4),
                createPage(5),
                createPage(6),
                createPage(7),
                createPage(8),
                createPage(9)));
        // page not acknowledged yet so sent count is still zero
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 11, 0, 11, 11);
        // acknowledge the 10 pages
        sharedBuffer.get(SECOND, 10, sizeOfPages(10)).cancel(true);
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 1, 10, 11, 11);

        //
        // tell shared buffer there will be no more queues
        outputBuffers = outputBuffers.withNoMoreBufferIds();
        sharedBuffer.setOutputBuffers(outputBuffers);

        // queues consumed the first three pages, so they should be dropped now and the blocked page future from above should be done
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 8, 3, 8, 11);
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 1, 10, 8, 11);
        assertFutureIsDone(future);

        // we should be able to add 3 more pages (the third will be queued)
        // although the first queue fetched the 4th page, the page has not been acknowledged yet
        addPage(sharedBuffer, createPage(11));
        addPage(sharedBuffer, createPage(12));
        future = enqueuePage(sharedBuffer, createPage(13));
        assertFalse(future.isDone());
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 11, 3, 11, 14);
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 4, 10, 11, 14);

        // acknowledge the receipt of the 3rd page and try to remove the 4th page from the first queue
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 4, sizeOfPages(1), NO_WAIT), bufferResult(4, createPage(4)));

        // the blocked page future above should be done
        assertFutureIsDone(future);
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 10, 4, 10, 14);
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 4, 10, 10, 14);

        //
        // finish the buffer
        assertFalse(sharedBuffer.isFinished());
        sharedBuffer.setNoMorePages();
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 10, 4, 10, 14);
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 4, 10, 10, 14);

        // not fully finished until all pages are consumed
        assertFalse(sharedBuffer.isFinished());

        // remove a page, not finished
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 5, sizeOfPages(1), NO_WAIT), bufferResult(5, createPage(5)));
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 9, 5, 9, 14);
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 4, 10, 9, 14);
        assertFalse(sharedBuffer.isFinished());

        // remove all remaining pages from first queue, should not be finished
        BufferResult x = getBufferResult(sharedBuffer, FIRST, 6, sizeOfPages(10), NO_WAIT);
        assertBufferResultEquals(TYPES, x, bufferResult(6, createPage(6),
                createPage(7),
                createPage(8),
                createPage(9),
                createPage(10),
                createPage(11),
                createPage(12),
                createPage(13)));
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 8, 6, 8, 14);
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 14, sizeOfPages(10), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 14, true));

        // finish first queue
        sharedBuffer.abort(FIRST);
        assertQueueClosed(sharedBuffer, FIRST, 14);
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 4, 10, 4, 14);
        assertFalse(sharedBuffer.isFinished());

        // remove all remaining pages from second queue, should be finished
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, SECOND, 10, sizeOfPages(10), NO_WAIT), bufferResult(10, createPage(10),
                createPage(11),
                createPage(12),
                createPage(13)));
        assertQueueState(sharedBuffer, SECOND, DEFAULT_PARTITION, 4, 10, 4, 14);
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, SECOND, 14, sizeOfPages(10), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 14, true));
        sharedBuffer.abort(SECOND);
        assertQueueClosed(sharedBuffer, FIRST, 14);
        assertQueueClosed(sharedBuffer, SECOND, 14);
        assertFinished(sharedBuffer);

        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 14, sizeOfPages(10), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 14, true));
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, SECOND, 14, sizeOfPages(10), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 14, true));
    }

    @Test
    public void testSharedBufferFull()
            throws Exception
    {
        int firstPartition = 0;
        int secondPartition = 1;
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(2));

        // Add two pages, buffer is full
        addPage(sharedBuffer, createPage(1), firstPartition);
        addPage(sharedBuffer, createPage(2), secondPartition);

        // third page is blocked
        enqueuePage(sharedBuffer, createPage(3), secondPartition);
    }

    @Test
    public void testDeqeueueOnAcknowledgement()
            throws Exception
    {
        int firstPartition = 0;
        int secondPartition = 1;
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(2));
        OutputBuffers outputBuffers = createInitialEmptyOutputBuffers(BROADCAST)
                .withBuffer(FIRST, firstPartition)
                .withBuffer(SECOND, secondPartition)
                .withNoMoreBufferIds();
        sharedBuffer.setOutputBuffers(outputBuffers);

        // Add two pages, buffer is full
        addPage(sharedBuffer, createPage(1), firstPartition);
        addPage(sharedBuffer, createPage(2), firstPartition);
        assertQueueState(sharedBuffer, FIRST, firstPartition, 2, 0, 2, 2);

        // third page is blocked
        ListenableFuture<?> future = enqueuePage(sharedBuffer, createPage(3), secondPartition);

        // we should be blocked
        assertFalse(future.isDone());
        assertQueueState(sharedBuffer, FIRST, firstPartition, 2, 0, 2, 2);   // 2 buffered pages
        assertQueueState(sharedBuffer, SECOND, secondPartition, 1, 0, 1, 1); // 1 queued page

        // acknowledge pages for first partition, make space in the shared buffer
        sharedBuffer.get(FIRST, 2, sizeOfPages(10)).cancel(true);

        // page should be dequeued, we should not be blocked
        assertFutureIsDone(future);
        assertQueueState(sharedBuffer, SECOND, secondPartition, 1, 0, 1, 1); // no more queued pages
    }

    @Test
    public void testSimplePartitioned()
            throws Exception
    {
        int firstPartition = 0;
        int secondPartition = 1;
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(20));

        // add three items to each buffer
        for (int i = 0; i < 3; i++) {
            addPage(sharedBuffer, createPage(i), firstPartition);
            addPage(sharedBuffer, createPage(i), secondPartition);
        }

        OutputBuffers outputBuffers = createInitialEmptyOutputBuffers(PARTITIONED).withBuffer(FIRST, firstPartition);

        // add first partition
        sharedBuffer.setOutputBuffers(outputBuffers);
        assertQueueState(sharedBuffer, FIRST, firstPartition, 3, 0, 3, 3);

        // get the three elements
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 0, sizeOfPages(10), NO_WAIT), bufferResult(0, createPage(0), createPage(1), createPage(2)));
        // pages not acknowledged yet so state is the same
        assertQueueState(sharedBuffer, FIRST, firstPartition, 3, 0, 3, 3);

        // acknowledge first three pages
        sharedBuffer.get(FIRST, 3, sizeOfPages(10)).cancel(true);
        // pages now acknowledged
        assertQueueState(sharedBuffer, FIRST, firstPartition, 0, 3, 3, 3);

        // fill the buffer (we already added 3 pages)
        for (int i = 3; i < 10; i++) {
            addPage(sharedBuffer, createPage(i), firstPartition);
            addPage(sharedBuffer, createPage(i), secondPartition);
        }
        assertQueueState(sharedBuffer, FIRST, firstPartition, 7, 3, 10, 10);

        // try to add one more page, which should block
        ListenableFuture<?> future = enqueuePage(sharedBuffer, createPage(10), firstPartition);
        assertFalse(future.isDone());
        assertQueueState(sharedBuffer, FIRST, firstPartition, 8, 3, 11, 11);

        // remove a page
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 3, sizeOfPages(1), NO_WAIT), bufferResult(3, createPage(3)));
        // page not acknowledged yet so sent count is the same
        assertQueueState(sharedBuffer, FIRST, firstPartition, 8, 3, 11, 11);

        // we should still be blocked
        assertFalse(future.isDone());

        //
        // add second partition and verify it sees all pages
        outputBuffers = outputBuffers.withBuffer(SECOND, secondPartition);
        sharedBuffer.setOutputBuffers(outputBuffers);
        assertQueueState(sharedBuffer, SECOND, secondPartition, 10, 0, 10, 10);
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, SECOND, 0, sizeOfPages(10), NO_WAIT), bufferResult(0, createPage(0),
                createPage(1),
                createPage(2),
                createPage(3),
                createPage(4),
                createPage(5),
                createPage(6),
                createPage(7),
                createPage(8),
                createPage(9)));
        // page not acknowledged yet so sent count is still zero
        assertQueueState(sharedBuffer, SECOND, secondPartition, 10, 0, 10, 10);
        // acknowledge the 10 pages
        sharedBuffer.get(SECOND, 10, sizeOfPages(10)).cancel(true);
        assertQueueState(sharedBuffer, SECOND, secondPartition, 0, 10, 10, 10);

        //
        // tell shared buffer there will be no more queues
        outputBuffers = outputBuffers.withNoMoreBufferIds();
        sharedBuffer.setOutputBuffers(outputBuffers);

        // since both queues consumed some pages, the blocked page future from above should be done
        assertFutureIsDone(future);
        assertQueueState(sharedBuffer, FIRST, firstPartition, 8, 3, 8, 11);
        assertQueueState(sharedBuffer, SECOND, secondPartition, 0, 10, 0, 10);

        // we should be able to add 3 more pages
        // although the first queue fetched the 4th page, the page has not been acknowledged yet
        addPage(sharedBuffer, createPage(11), firstPartition);
        addPage(sharedBuffer, createPage(12), firstPartition);
        addPage(sharedBuffer, createPage(13), firstPartition);
        assertQueueState(sharedBuffer, FIRST, firstPartition, 11, 3, 11, 14);
        assertQueueState(sharedBuffer, SECOND, secondPartition, 0, 10, 0, 10);

        // remove a page from the first queue
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 4, sizeOfPages(1), NO_WAIT), bufferResult(4, createPage(4)));

        // the blocked page future above should be done
        assertFutureIsDone(future);
        assertQueueState(sharedBuffer, FIRST, firstPartition, 10, 4, 10, 14);
        assertQueueState(sharedBuffer, SECOND, secondPartition, 0, 10, 0, 10);

        //
        // finish the buffer
        assertFalse(sharedBuffer.isFinished());
        sharedBuffer.setNoMorePages();
        assertQueueState(sharedBuffer, FIRST, firstPartition, 10, 4, 10, 14);
        assertQueueState(sharedBuffer, SECOND, secondPartition, 0, 10, 0, 10);
        sharedBuffer.abort(SECOND);
        assertQueueClosed(sharedBuffer, SECOND, 10);

        // not fully finished until all pages are consumed
        assertFalse(sharedBuffer.isFinished());

        // remove a page, not finished
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 5, sizeOfPages(1), NO_WAIT), bufferResult(5, createPage(5)));
        assertQueueState(sharedBuffer, FIRST, firstPartition, 9, 5, 9, 14);
        assertFalse(sharedBuffer.isFinished());

        // remove all remaining pages from first queue, should not be finished
        BufferResult x = getBufferResult(sharedBuffer, FIRST, 6, sizeOfPages(10), NO_WAIT);
        assertBufferResultEquals(TYPES, x, bufferResult(6, createPage(6),
                createPage(7),
                createPage(8),
                createPage(9),
                createPage(10),
                createPage(11),
                createPage(12),
                createPage(13)));
        assertQueueState(sharedBuffer, FIRST, firstPartition, 8, 6, 8, 14);
        // acknowledge all pages from the first partition, should transition to finished state
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 14, sizeOfPages(10), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 14, true));
        sharedBuffer.abort(FIRST);
        assertQueueClosed(sharedBuffer, FIRST, 14);
        assertFinished(sharedBuffer);
    }

    public static BufferResult getBufferResult(SharedOutputBuffer sharedBuffer, OutputBufferId bufferId, long sequenceId, DataSize maxSize, Duration maxWait)
    {
        CompletableFuture<BufferResult> future = sharedBuffer.get(bufferId, sequenceId, maxSize);
        return getFuture(future, maxWait);
    }

    public static BufferResult getFuture(CompletableFuture<BufferResult> future, Duration maxWait)
    {
        return tryGetFutureValue(future, (int) maxWait.toMillis(), TimeUnit.MILLISECONDS).get();
    }

    @Test
    public void testDuplicateRequests()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));
        // add three items
        for (int i = 0; i < 3; i++) {
            addPage(sharedBuffer, createPage(i));
        }

        // add a queue
        OutputBuffers outputBuffers = createInitialEmptyOutputBuffers(BROADCAST);
        outputBuffers = outputBuffers.withBuffer(FIRST, 0);
        sharedBuffer.setOutputBuffers(outputBuffers);
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 3, 0, 3, 3);

        // get the three elements
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 0, sizeOfPages(10), NO_WAIT), bufferResult(0, createPage(0), createPage(1), createPage(2)));
        // pages not acknowledged yet so state is the same
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 3, 0, 3, 3);

        // get the three elements again
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 0, sizeOfPages(10), NO_WAIT), bufferResult(0, createPage(0), createPage(1), createPage(2)));
        // pages not acknowledged yet so state is the same
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 3, 0, 3, 3);

        // acknowledge the pages
        sharedBuffer.get(FIRST, 3, sizeOfPages(10)).cancel(true);

        // attempt to get the three elements again
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 0, sizeOfPages(10), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 0, false));
        // pages not acknowledged yet so state is the same
        assertQueueState(sharedBuffer, FIRST, DEFAULT_PARTITION, 0, 3, 3, 3);
    }

    @Test
    public void testAddQueueAfterNoMoreQueues()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));
        assertFalse(sharedBuffer.isFinished());

        // tell buffer no more queues will be added
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withNoMoreBufferIds());
        assertFalse(sharedBuffer.isFinished());

        // set no more queues a second time to assure that we don't get an exception or such
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withNoMoreBufferIds());
        assertFalse(sharedBuffer.isFinished());

        // set no more queues a third time to assure that we don't get an exception or such
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withNoMoreBufferIds());
        assertFalse(sharedBuffer.isFinished());

        try {
            OutputBuffers outputBuffers = createInitialEmptyOutputBuffers(BROADCAST)
                    .withBuffer(FOO, 0)
                    .withNoMoreBufferIds();

            sharedBuffer.setOutputBuffers(outputBuffers);
            fail("Expected IllegalStateException from addQueue after noMoreQueues has been called");
        }
        catch (IllegalArgumentException ignored) {
        }
    }

    @Test
    public void testAddQueueAfterDestroy()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));
        assertFalse(sharedBuffer.isFinished());

        // destroy buffer
        sharedBuffer.destroy();
        assertFinished(sharedBuffer);

        // set no more queues to assure that we don't get an exception or such
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withNoMoreBufferIds());
        assertFinished(sharedBuffer);

        // set no more queues a second time to assure that we don't get an exception or such
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withNoMoreBufferIds());
        assertFinished(sharedBuffer);

        // add queue calls after finish should be ignored
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withBuffer(FOO, 0).withNoMoreBufferIds());
    }

    @Test
    public void testGetBeforeCreate()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));
        assertFalse(sharedBuffer.isFinished());

        // get a page from a buffer that doesn't exist yet
        CompletableFuture<BufferResult> future = sharedBuffer.get(FIRST, (long) 0, sizeOfPages(1));
        assertFalse(future.isDone());

        // add a page and verify the future is not complete
        addPage(sharedBuffer, createPage(33));
        assertFalse(future.isDone());

        // add the buffer and verify the future completed
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withBuffer(FIRST, 0));
        assertBufferResultEquals(TYPES, getFuture(future, NO_WAIT), bufferResult(0, createPage(33)));
    }

    @Test
    public void testAbortBeforeCreate()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));
        assertFalse(sharedBuffer.isFinished());

        // get a page from a buffer that doesn't exist yet
        CompletableFuture<BufferResult> future = sharedBuffer.get(FIRST, (long) 0, sizeOfPages(1));
        assertFalse(future.isDone());

        // abort that buffer
        sharedBuffer.abort(FIRST);

        // add a page and verify the future is not complete
        addPage(sharedBuffer, createPage(33));
        assertFalse(future.isDone());

        // add the buffer and verify we did not get the page
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withBuffer(FIRST, 0));
        assertBufferResultEquals(TYPES, getFuture(future, NO_WAIT), emptyResults(TASK_INSTANCE_ID, 0, true));

        // verify that a normal read returns a closed empty result
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 0, sizeOfPages(10), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 0, true));
    }

    @Test
    public void testAddStateMachine()
            throws Exception
    {
        // add after finish
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));
        sharedBuffer.setNoMorePages();
        addPage(sharedBuffer, createPage(0));
        addPage(sharedBuffer, createPage(0));
        assertEquals(sharedBuffer.getInfo().getTotalPagesSent(), 0);

        // add after destroy
        sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));
        sharedBuffer.destroy();
        addPage(sharedBuffer, createPage(0));
        addPage(sharedBuffer, createPage(0));
        assertEquals(sharedBuffer.getInfo().getTotalPagesSent(), 0);
    }

    @Test
    public void testAbort()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));

        // fill the buffer
        for (int i = 0; i < 10; i++) {
            addPage(sharedBuffer, createPage(i));
        }
        sharedBuffer.setNoMorePages();

        OutputBuffers outputBuffers = createInitialEmptyOutputBuffers(BROADCAST)
                .withBuffer(FIRST, 0);
        sharedBuffer.setOutputBuffers(outputBuffers);
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 0, sizeOfPages(1), NO_WAIT), bufferResult(0, createPage(0)));
        sharedBuffer.abort(FIRST);
        assertQueueClosed(sharedBuffer, FIRST, 0);
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, FIRST, 1, sizeOfPages(1), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 1, true));

        outputBuffers = outputBuffers.withBuffer(SECOND, 0).withNoMoreBufferIds();
        sharedBuffer.setOutputBuffers(outputBuffers);
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, SECOND, 0, sizeOfPages(1), NO_WAIT), bufferResult(0, createPage(0)));
        sharedBuffer.abort(SECOND);
        assertQueueClosed(sharedBuffer, SECOND, 0);
        assertFinished(sharedBuffer);
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, SECOND, 1, sizeOfPages(1), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 0, true));
    }

    @Test
    public void testFinishClosesEmptyQueues()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(10));
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST)
                .withBuffer(FIRST, 0)
                .withBuffer(SECOND, 0));

        // finish while queues are empty
        sharedBuffer.setNoMorePages();

        assertQueueState(sharedBuffer, FIRST, 0, 0, 0, 0, 0);
        assertQueueState(sharedBuffer, SECOND, 0, 0, 0, 0, 0);

        sharedBuffer.abort(FIRST);
        sharedBuffer.abort(SECOND);

        assertQueueClosed(sharedBuffer, FIRST, 0);
        assertQueueClosed(sharedBuffer, SECOND, 0);
    }

    @Test
    public void testAbortFreesReader()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(5));
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withBuffer(QUEUE, 0));
        assertFalse(sharedBuffer.isFinished());

        // attempt to get a page
        CompletableFuture<BufferResult> future = sharedBuffer.get(QUEUE, 0, sizeOfPages(10));

        // verify we are waiting for a page
        assertFalse(future.isDone());

        // add one item
        addPage(sharedBuffer, createPage(0));

        // verify we got one page
        assertBufferResultEquals(TYPES, getFuture(future, NO_WAIT), bufferResult(0, createPage(0)));

        // attempt to get another page, and verify we are blocked
        future = sharedBuffer.get(QUEUE, 1, sizeOfPages(10));
        assertFalse(future.isDone());

        // abort the buffer
        sharedBuffer.abort(QUEUE);
        assertQueueClosed(sharedBuffer, QUEUE, 1);

        // verify the future completed
        assertBufferResultEquals(TYPES, getFuture(future, NO_WAIT), emptyResults(TASK_INSTANCE_ID, 1, true));
    }

    @Test
    public void testFinishFreesReader()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(5));
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST).withBuffer(QUEUE, 0));
        assertFalse(sharedBuffer.isFinished());

        // attempt to get a page
        CompletableFuture<BufferResult> future = sharedBuffer.get(QUEUE, 0, sizeOfPages(10));

        // verify we are waiting for a page
        assertFalse(future.isDone());

        // add one item
        addPage(sharedBuffer, createPage(0));

        // verify we got one page
        assertBufferResultEquals(TYPES, getFuture(future, NO_WAIT), bufferResult(0, createPage(0)));

        // attempt to get another page, and verify we are blocked
        future = sharedBuffer.get(QUEUE, 1, sizeOfPages(10));
        assertFalse(future.isDone());

        // finish the buffer
        sharedBuffer.setNoMorePages();
        assertQueueState(sharedBuffer, QUEUE, 0, 0, 1, 1, 1);
        sharedBuffer.abort(QUEUE);
        assertQueueClosed(sharedBuffer, QUEUE, 1);

        // verify the future completed
        assertBufferResultEquals(TYPES, getFuture(future, NO_WAIT), emptyResults(TASK_INSTANCE_ID, 1, true));
    }

    @Test
    public void testFinishFreesWriter()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(5));
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST)
                .withBuffer(QUEUE, 0)
                .withNoMoreBufferIds());
        assertFalse(sharedBuffer.isFinished());

        // fill the buffer
        for (int i = 0; i < 5; i++) {
            addPage(sharedBuffer, createPage(i));
        }

        // enqueue the addition two pages more pages
        ListenableFuture<?> firstEnqueuePage = enqueuePage(sharedBuffer, createPage(5));
        ListenableFuture<?> secondEnqueuePage = enqueuePage(sharedBuffer, createPage(6));

        // get and acknowledge one page
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, QUEUE, 0, sizeOfPages(1), MAX_WAIT), bufferResult(0, createPage(0)));
        sharedBuffer.get(QUEUE, 1, sizeOfPages(1)).cancel(true);

        // verify we are still blocked because the buffer is full
        assertFalse(firstEnqueuePage.isDone());
        assertFalse(secondEnqueuePage.isDone());

        // finish the query
        sharedBuffer.setNoMorePages();
        assertFalse(sharedBuffer.isFinished());

        // verify futures are complete
        assertFutureIsDone(firstEnqueuePage);
        assertFutureIsDone(secondEnqueuePage);

        // get and acknowledge the last 6 pages
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, QUEUE, 1, sizeOfPages(100), NO_WAIT),
                bufferResult(1, createPage(1), createPage(2), createPage(3), createPage(4), createPage(5), createPage(6)));
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, QUEUE, 7, sizeOfPages(100), NO_WAIT), emptyResults(TASK_INSTANCE_ID, 7, true));

        sharedBuffer.abort(QUEUE);

        // verify finished
        assertFinished(sharedBuffer);
    }

    @Test
    public void testDestroyFreesReader()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(5));
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST)
                .withBuffer(QUEUE, 0)
                .withNoMoreBufferIds());
        assertFalse(sharedBuffer.isFinished());

        // attempt to get a page
        CompletableFuture<BufferResult> future = sharedBuffer.get(QUEUE, 0, sizeOfPages(10));

        // verify we are waiting for a page
        assertFalse(future.isDone());

        // add one page
        addPage(sharedBuffer, createPage(0));

        // verify we got one page
        assertBufferResultEquals(TYPES, getFuture(future, NO_WAIT), bufferResult(0, createPage(0)));

        // attempt to get another page, and verify we are blocked
        future = sharedBuffer.get(QUEUE, 1, sizeOfPages(10));
        assertFalse(future.isDone());

        // destroy the buffer
        sharedBuffer.destroy();
        assertQueueClosed(sharedBuffer, QUEUE, 1);

        // verify the future completed
        assertBufferResultEquals(TYPES, getFuture(future, NO_WAIT), emptyResults(TASK_INSTANCE_ID, 1, true));
    }

    @Test
    public void testDestroyFreesWriter()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(5));
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST)
                .withBuffer(QUEUE, 0)
                .withNoMoreBufferIds());
        assertFalse(sharedBuffer.isFinished());

        // fill the buffer
        for (int i = 0; i < 5; i++) {
            addPage(sharedBuffer, createPage(i));
        }

        // add two pages to the buffer queue
        ListenableFuture<?> firstEnqueuePage = enqueuePage(sharedBuffer, createPage(5));
        ListenableFuture<?> secondEnqueuePage = enqueuePage(sharedBuffer, createPage(6));

        // get and acknowledge one page
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, QUEUE, 0, sizeOfPages(1), MAX_WAIT), bufferResult(0, createPage(0)));
        sharedBuffer.get(QUEUE, 1, sizeOfPages(1)).cancel(true);

        // verify we are still blocked because the buffer is full
        assertFalse(firstEnqueuePage.isDone());
        assertFalse(secondEnqueuePage.isDone());

        // destroy the buffer (i.e., cancel the query)
        sharedBuffer.destroy();
        assertFinished(sharedBuffer);

        // verify the futures are completed
        assertFutureIsDone(firstEnqueuePage);
        assertFutureIsDone(secondEnqueuePage);
    }

    @Test
    public void testBufferCompletion()
            throws Exception
    {
        SharedOutputBuffer sharedBuffer = new SharedOutputBuffer(TASK_ID, TASK_INSTANCE_ID, stateNotificationExecutor, retainedSizeOfPages(5));
        sharedBuffer.setOutputBuffers(createInitialEmptyOutputBuffers(BROADCAST)
                .withBuffer(QUEUE, 0)
                .withNoMoreBufferIds());

        assertFalse(sharedBuffer.isFinished());

        // fill the buffer
        List<Page> pages = new ArrayList<>();
        for (int i = 0; i < 5; i++) {
            Page page = createPage(i);
            addPage(sharedBuffer, page);
            pages.add(page);
        }

        sharedBuffer.setNoMorePages();

        // get and acknowledge 5 pages
        assertBufferResultEquals(TYPES, getBufferResult(sharedBuffer, QUEUE, 0, sizeOfPages(5), MAX_WAIT), bufferResult(0, pages));

        // buffer is not finished
        assertFalse(sharedBuffer.isFinished());

        // there are no more pages and no more buffers, but buffer is not finished because it didn't receive an acknowledgement yet
        assertFalse(sharedBuffer.isFinished());

        // ask the buffer to finish
        sharedBuffer.abort(QUEUE);

        // verify that the buffer is finished
        assertTrue(sharedBuffer.isFinished());
    }

    private static ListenableFuture<?> enqueuePage(SharedOutputBuffer sharedBuffer, Page page)
    {
        return enqueuePage(sharedBuffer, page, DEFAULT_PARTITION);
    }

    private static ListenableFuture<?> enqueuePage(SharedOutputBuffer sharedBuffer, Page page, int partition)
    {
        ListenableFuture<?> future = sharedBuffer.enqueue(partition, page);
        assertFalse(future.isDone());
        return future;
    }

    private static void addPage(SharedOutputBuffer sharedBuffer, Page page)
    {
        addPage(sharedBuffer, page, DEFAULT_PARTITION);
    }

    private static void addPage(SharedOutputBuffer sharedBuffer, Page page, int partition)
    {
        assertTrue(sharedBuffer.enqueue(partition, page).isDone());
    }

    private static void assertQueueState(
            SharedOutputBuffer sharedBuffer,
            OutputBufferId bufferId,
            int partition,
            int bufferedPages,
            int pagesSent,
            int pageBufferBufferedPages,
            int pageBufferPagesSent)
    {
        assertEquals(
                getBufferInfo(sharedBuffer, bufferId),
                new BufferInfo(
                        bufferId,
                        false,
                        bufferedPages,
                        pagesSent,
                        new PageBufferInfo(
                                partition,
                                pageBufferBufferedPages,
                                retainedSizeOfPages(pageBufferBufferedPages).toBytes(),
                                pageBufferPagesSent, // every page has one row
                                pageBufferPagesSent)));
    }

    @SuppressWarnings("ConstantConditions")
    private static void assertQueueClosed(SharedOutputBuffer sharedBuffer, OutputBufferId bufferId, int pagesSent)
    {
        BufferInfo bufferInfo = getBufferInfo(sharedBuffer, bufferId);
        assertEquals(bufferInfo.getBufferedPages(), 0);
        assertEquals(bufferInfo.getPagesSent(), pagesSent);
        assertEquals(bufferInfo.isFinished(), true);
    }

    private static BufferInfo getBufferInfo(SharedOutputBuffer sharedBuffer, OutputBufferId bufferId)
    {
        for (BufferInfo bufferInfo : sharedBuffer.getInfo().getBuffers()) {
            if (bufferInfo.getBufferId().equals(bufferId)) {
                return bufferInfo;
            }
        }
        return null;
    }

    private static void assertFinished(SharedOutputBuffer sharedBuffer)
            throws Exception
    {
        assertTrue(sharedBuffer.isFinished());
        for (BufferInfo bufferInfo : sharedBuffer.getInfo().getBuffers()) {
            assertTrue(bufferInfo.isFinished());
            assertEquals(bufferInfo.getBufferedPages(), 0);
        }
    }

    private static void assertBufferResultEquals(List<? extends Type> types, BufferResult actual, BufferResult expected)
    {
        assertEquals(actual.getPages().size(), expected.getPages().size());
        assertEquals(actual.getToken(), expected.getToken());
        for (int i = 0; i < actual.getPages().size(); i++) {
            Page actualPage = actual.getPages().get(i);
            Page expectedPage = expected.getPages().get(i);
            assertEquals(actualPage.getChannelCount(), expectedPage.getChannelCount());
            PageAssertions.assertPageEquals(types, actualPage, expectedPage);
        }
        assertEquals(actual.isBufferComplete(), expected.isBufferComplete());
    }

    private static void assertFutureIsDone(Future<?> future)
    {
        tryGetFutureValue(future, 5, SECONDS);
        assertTrue(future.isDone());
    }

    public static BufferResult bufferResult(long token, Page firstPage, Page... otherPages)
    {
        List<Page> pages = ImmutableList.<Page>builder().add(firstPage).add(otherPages).build();
        return bufferResult(token, pages);
    }

    public static BufferResult bufferResult(long token, List<Page> pages)
    {
        checkArgument(!pages.isEmpty(), "pages is empty");
        return new BufferResult(TASK_INSTANCE_ID, token, token + pages.size(), false, pages);
    }
}


text:
package com.kotcrab.vis.runtime.spriter;

/**
 * An Exception which will be thrown if a Spriter specific issue happens at runtime.
 * @author Trixt0r
 */
public class SpriterException extends RuntimeException {

	private static final long serialVersionUID = 1L;

	public SpriterException (String message) {
		super(message);
	}

}

redpajama_set_name: RedPajamaGithub


text:
package iface

import "strconv"

type Proto int

const (
	TCP Proto = iota
	UDP
	HTTP
)

func (p Proto) String() string {
	switch p {
	case TCP:
		return "tcp"
	case UDP:
		return "udp"
	case HTTP:
		return "http"
	default:
		return "unknown"
	}
}

type Address struct {
	IP   string
	Port int
}

func (a Address) String() string {
	return a.IP + ":" + strconv.Itoa(a.Port)
}

type Net interface {
	Get(addr Address, path string) ([]byte, error)
	Put(addr Address, path string, data []byte) error
	Post(addr Address, path string, data []byte) ([]byte, error)
	Listen(addr Address, handler func(string, []byte) []byte, shutdown chan bool) error
}


text:
'use strict';

const gulp            = require('gulp'),
			sass            = require('gulp-sass'),
			prefix          = require('gulp-autoprefixer'),
			sourcemaps      = require('gulp-sourcemaps'),
			plumber         = require('gulp-plumber'),
			gutil           = require('gulp-util');

const PRODUCTION = gutil.env.production;

gulp.task('styles', () => {
	return gulp.src('./src/assets/scss/*.scss')
		.pipe(plumber())
		.pipe(sourcemaps.init())
		.pipe(sass({
			includePaths: [
				'./node_modules/foundation-sites/scss'
			]
		}))
		.pipe(sass())
		.pipe(prefix({
			browsers: ['last 2 versions'],
			cascade: false
		}))
		// .pipe(sass({outputStyle: 'compressed'}))
		.pipe(PRODUCTION ? sass({outputStyle: 'compressed'}) : gutil.noop())
		.pipe(sourcemaps.write('.'))
		.pipe(gulp.dest('./dist/assets/css'));
});


text:
using Microsoft.AspNetCore.Mvc.RazorPages;
using Microsoft.Extensions.Logging;

namespace NetFiveSimple.Pages
{
    public class IndexModel : PageModel
    {
        private readonly ILogger<IndexModel> _logger;

        public IndexModel(ILogger<IndexModel> logger)
        {
            _logger = logger;
        }

        public void OnGet()
        {

        }
    }
}


text:
package org.openengsb.core.workflow.drools;

/**
 * is thrown when the service backing an {@link org.openengsb.core.common.communication.IncomingPort} or
 * {@link org.openengsb.core.common.communication.OutgoingPort} is not available
 */
@SuppressWarnings("serial")
public class PortNotAvailableException extends Exception {

    public PortNotAvailableException() {
        super();
    }

    public PortNotAvailableException(String message, Throwable cause) {
        super(message, cause);
    }

    public PortNotAvailableException(String message) {
        super(message);
    }

    public PortNotAvailableException(Throwable cause) {
        super(cause);
    }

}


text:
<?php

declare(strict_types=1);

/*
 * (c) Jeroen van den Enden <info@endroid.nl>
 *
 * This source file is subject to the MIT license that is bundled
 * with this source code in the file LICENSE.
 */

namespace Endroid\QrCode\Exception;

class ValidationException extends QrCodeException
{
}


text:
package main

import (
	"io"

	"github.com/spf13/cobra"
	"k8s.io/kops/cmd/kops/util"
	"k8s.io/kubernetes/pkg/kubectl/cmd/templates"
	"k8s.io/kubernetes/pkg/kubectl/util/i18n"
)

var (
	setLong = templates.LongDesc(i18n.T(`Set a configuration field.

        kops set does not update the cloud resources, to apply the changes use "kops update cluster".
    `))

	setExample = templates.Examples(i18n.T(`
    # Set cluster to run kubernetes version 1.10.0
    kops set cluster k8s-cluster.example.com spec.kubernetesVersion=1.10.0
	`))
)

func NewCmdSet(f *util.Factory, out io.Writer) *cobra.Command {
	cmd := &cobra.Command{
		Use:     "set",
		Short:   i18n.T("Set fields on clusters and other resources."),
		Long:    setLong,
		Example: setExample,
	}

	// create subcommands
	cmd.AddCommand(NewCmdSetCluster(f, out))

	return cmd
}


text:
<manifest xmlns:android="http://schemas.android.com/apk/res/android"
    package="co.geekku.lib">

    <application android:allowBackup="true"
        android:label="@string/app_name">

    </application>

</manifest>


text: Rails.application.config.session_store :cookie_store, key: '_bank_session'


text:
import Ember from 'ember';

export default Ember.Route.extend({

  model() {
    return this.store.findAll('contact')
      .catch((reason) => {
        var errorMsg = reason.responseJSON ? reason.responseJSON.errors[0] : reason.errors[0];

        this.set('error', errorMsg);
      });
  },

  setupController(controller, model) {
    if (this.get('error')) {
      controller.set('error', this.get('error'));
    } else {
      controller.set('model', model);
    }
  }

});


text:
import ObjectUtil from './ObjectUtil';

/**
 * Provides common string utilities.
 */
export default class StringUtil
{
   /**
    * Compares two strings.
    *
    * @param {string}   lhs - Left-hand side.
    * @param {string}   rhs - Right-hand side.
    *
    * @returns {number}
    */
   static compare(lhs, rhs)
   {
      return lhs.toLowerCase().localeCompare(rhs.toLowerCase());
   }

   /**
    * Increments the indentation amount.
    *
    * @param {number}   indentation - Current indentation amount.
    * @param {number}   amount - (Optional) indentation amount; default: 3.
    *
    * @returns {number}
    */
   static incrementIndent(indentation, amount = 3)
   {
      return indentation + amount;
   }

   /**
    * Creates an indentation string given the indentation amount.
    *
    * @param {number}   indentation - Current indentation amount.
    * @param {string}   string - A string to append.
    *
    * @returns {string}
    */
   static indent(indentation, string = '')
   {
      return (new Array(indentation + 1)).join(' ') + string;
   }

   /**
    * Returns the SafeEntry constructor which is used by `safeStringsObject` and `safeStringsPrependObject`.
    *
    * @returns {SafeEntry}
    */
   static get SafeEntry() { return SafeEntry; }

   /**
    * Provides a way to output a given string value with concatenated data from safely accessing an objects data /
    * entries given an accessor string which describes the entries to walk. To access deeper entries into the object
    * format the accessor string with `.` between entries to walk.
    *
    * @param {string}   string - A string to prepend to the object data received.
    * @param {object}   object - An object to access entry data.
    * @param {string}   accessor - A string describing the entries to access.
    * @param {number}   newLine - (Optional) A number of new line characters to append; default: `1`.
    * @param {string}   appendString - (Optional) A string to potentially append; default: `''`;
    * @param {function} tagFunction - (Optional) A template tag function to apply; default: `void 0`;
    *
    * @returns {string}
    */
   static safeStringObject(string, object, accessor, newLine = 1, appendString = '', tagFunction = void 0)
   {
      const value = ObjectUtil.safeAccess(object, accessor);

      if (typeof value === 'undefined' || (Array.isArray(value) && value.length === 0)) { return ''; }

      let end = '\n';

      // Create the ending new line result if it is not the default of `1`.
      if (newLine === 0 || newLine > 1) { end = new Array(newLine + 1).join('\n'); }

      return typeof tagFunction === 'function' ? tagFunction`${string}${value}${appendString}${end}` :
       `${string}${value}${appendString}${end}`;
   }

   /**
    * Provides a convenience method producing a block of `safeStringObject` results.
    *
    * @param {object}         object - An object to access entry data.
    *
    * @param {string|SafeEntry|Array<string|SafeEntry>} entries -
    *                                  Multiple arrays or strings. If an entry is not a SafeEntry it will
    *                                  simply be appended. If the entry is an array then entries in this array
    *                                  correspond to the following parameters which are forwarded onto
    *                                  `safeStringObject`.
    *
    * @returns {string}
    */
   static safeStringsObject(object, ...entries)
   {
      return StringUtil.safeStringsPrependObject('', object, ...entries);
   }

   /**
    * Provides a convenience method producing a block of `safeStringObject` results with the option to prepend a
    * given string.
    *
    * @param {*}              origPrepend - An additional value to prepend to all results.
    *
    * @param {object}         object - An object to access entry data.
    *
    * @param {string|SafeEntry|Array<string|SafeEntry>} entries -
    *                                  Multiple arrays or strings. If an entry is not a SafeEntry it will
    *                                  simply be appended. If the entry is an array then entries in this array
    *                                  correspond to the following parameters which are forwarded onto
    *                                  `safeStringObject`.
    *
    * @returns {string}
    */
   static safeStringsPrependObject(origPrepend, object, ...entries)
   {
      if (typeof object !== 'object') { return ''; }

      const output = [];

      let skipPrepend = false;

      for (let cntr = 0; cntr < entries.length; cntr++)
      {
         const entry = entries[cntr];

         // Skip prepend action if last entry did not include a new line.
         const prepend = skipPrepend ? '' : origPrepend;

         // Process an array entry otherwise simply append `entry` to output if it is a string.
         if (entry instanceof SafeEntry)
         {
            skipPrepend = entry.newLine === 0;

            output.push(StringUtil.safeStringObject(`${prepend}${entry.prependString}`,
             object, entry.accessor, entry.newLine, entry.appendString, entry.tagFunction));
         }
         else if (typeof entry === 'string' && entry !== '')
         {
            output.push(`${prepend}${entry}`);
         }
         else if (typeof entry.toString === 'function')
         {
            const stringValue = entry.toString();

            if (stringValue !== '')
            {
               output.push(`${prepend}${stringValue}`);
            }
         }
      }

      return output.join('');
   }

   /**
    * Provides a tagged template method to escape HTML elements.
    *
    * @param {Array<string>}  literal - Literal components of template string.
    * @param {Array<*>}       values - Values to substitute.
    *
    * @returns {string}
    */
   static tagEscapeHTML(literal, ...values)
   {
      return values.reduce((previous, value, index) =>
      {
         return previous + String(value).replace(/</g, '<').replace(/>/g, '>') + literal[index + 1];
      }, literal[0]);
   }
}

/**
 * Defines the parameters for a safe string object lookup. If the accessor resolves against a given object then
 * a string is created by combining the prepend / append strings between the value with optional new lines appended
 * at the end. If a template tag function is defined it is applied to the template string.
 */
class SafeEntry
{
   /**
    * Initializes SafeEntry instance.
    *
    * @param {string}   prependString - The string to prepend.
    * @param {string}   accessor - The accessor string describing the lookup operation.
    * @param {number}   newLine - (Optional) The number of newlines characters to append.
    * @param {string}   appendString - (Optional) A string to append to the end.
    * @param {function} tagFunction - (Optional) A template tag function to apply.
    */
   constructor(prependString, accessor, newLine = 1, appendString = '', tagFunction = void 0)
   {
      if (typeof prependString !== 'string') { throw new TypeError(`ctor error: 'prependString' is not a 'string'.`); }
      if (typeof accessor !== 'string') { throw new TypeError(`ctor error: 'accessor' is not a 'string'.`); }
      if (typeof appendString !== 'string') { throw new TypeError(`ctor error: 'appendString' is not a 'string'.`); }

      if (typeof tagFunction !== 'function' && typeof tagFunction !== 'undefined')
      {
         throw new TypeError(`ctor error: 'tagFunction' is not a 'function' or 'undefined'.`);
      }

      if (Number.isInteger(newLine) && newLine < 0)
      {
         throw new TypeError(`ctor error: 'newLine' is not a positive 'integer' (${newLine}).`);
      }

      /**
       * The string to prepend.
       * @type {string}
       * @private
       */
      this._prependString = prependString;

      /**
       * The accessor string describing the lookup operation.
       * @type {string}
       * @private
       */
      this._accessor = accessor;

      /**
       * The number of newlines characters to append.
       * @type {number}
       * @private
       */
      this._newLine = newLine;

      /**
       * A string to append to the end.
       * @type {string}
       * @private
       */
      this._appendString = appendString;

      /**
       * A template tag function to apply.
       * @type {Function}
       * @private
       */
      this._tagFunction = tagFunction;
   }

   /**
    * Returns the accessor string describing the lookup operation.
    * @returns {string}
    */
   get accessor() { return this._accessor; }

   /**
    * Returns a string to append to the end if any.
    * @returns {string}
    */
   get appendString() { return this._appendString; }

   /**
    * Returns the new line count.
    * @returns {number}
    */
   get newLine() { return this._newLine; }

   /**
    * Returns the string to prepend.
    * @returns {string}
    */
   get prependString() { return this._prependString; }

   /**
    * Returns a template tag function to apply if any.
    * @returns {Function}
    */
   get tagFunction() { return this._tagFunction; }
}


text:
import React, { Component } from 'react';
import {
  AppRegistry,
  StyleSheet,
  Text,
  View
} from 'react-native';

import firebase from 'firebase';

class SignInPage extends Component {

  render() {
    return (

    );
  }
}
export default Home;


text:
import {
  BUILDER_KEYS,
  DEPRECATED_KEYS,
  NODE_FIELDS,
  toBindingIdentifierName,
} from "../../lib/index.js";
import formatBuilderName from "../utils/formatBuilderName.js";
import lowerFirst from "../utils/lowerFirst.js";
import stringifyValidator from "../utils/stringifyValidator.js";

function areAllRemainingFieldsNullable(fieldName, fieldNames, fields) {
  const index = fieldNames.indexOf(fieldName);
  return fieldNames.slice(index).every(_ => isNullable(fields[_]));
}

function hasDefault(field) {
  return field.default != null;
}

function isNullable(field) {
  return field.optional || hasDefault(field);
}

function sortFieldNames(fields, type) {
  return fields.sort((fieldA, fieldB) => {
    const indexA = BUILDER_KEYS[type].indexOf(fieldA);
    const indexB = BUILDER_KEYS[type].indexOf(fieldB);
    if (indexA === indexB) return fieldA < fieldB ? -1 : 1;
    if (indexA === -1) return 1;
    if (indexB === -1) return -1;
    return indexA - indexB;
  });
}

function generateBuilderArgs(type) {
  const fields = NODE_FIELDS[type];
  const fieldNames = sortFieldNames(Object.keys(NODE_FIELDS[type]), type);
  const builderNames = BUILDER_KEYS[type];

  const args = [];

  fieldNames.forEach(fieldName => {
    const field = fields[fieldName];
    // Future / annoying TODO:
    // MemberExpression.property, ObjectProperty.key and ObjectMethod.key need special cases; either:
    // - convert the declaration to chain() like ClassProperty.key and ClassMethod.key,
    // - declare an alias type for valid keys, detect the case and reuse it here,
    // - declare a disjoint union with, for example, ObjectPropertyBase,
    //   ObjectPropertyLiteralKey and ObjectPropertyComputedKey, and declare ObjectProperty
    //   as "ObjectPropertyBase & (ObjectPropertyLiteralKey | ObjectPropertyComputedKey)"
    let typeAnnotation = stringifyValidator(field.validate, "t.");

    if (isNullable(field) && !hasDefault(field)) {
      typeAnnotation += " | null";
    }

    if (builderNames.includes(fieldName)) {
      const field = NODE_FIELDS[type][fieldName];
      const def = JSON.stringify(field.default);
      const bindingIdentifierName = toBindingIdentifierName(fieldName);
      let arg;
      if (areAllRemainingFieldsNullable(fieldName, builderNames, fields)) {
        arg = `${bindingIdentifierName}${
          isNullable(field) && !def ? "?:" : ":"
        } ${typeAnnotation}`;
      } else {
        arg = `${bindingIdentifierName}: ${typeAnnotation}${
          isNullable(field) ? " | undefined" : ""
        }`;
      }
      if (def !== "null" || isNullable(field)) {
        arg += `= ${def}`;
      }
      args.push(arg);
    }
  });

  return args;
}

export default function generateBuilders(kind) {
  return kind === "uppercase.js"
    ? generateUppercaseBuilders()
    : generateLowercaseBuilders();
}

function generateLowercaseBuilders() {
  let output = `/*
 * This file is auto-generated! Do not modify it directly.
 * To re-generate run 'make build'
 */
import validateNode from "../validateNode";
import type * as t from "../..";
`;

  const reservedNames = new Set(["super", "import"]);
  Object.keys(BUILDER_KEYS).forEach(type => {
    const defArgs = generateBuilderArgs(type);
    const formatedBuilderName = formatBuilderName(type);
    const formatedBuilderNameLocal = reservedNames.has(formatedBuilderName)
      ? `_${formatedBuilderName}`
      : formatedBuilderName;

    const fieldNames = sortFieldNames(Object.keys(NODE_FIELDS[type]), type);
    const builderNames = BUILDER_KEYS[type];
    const objectFields = [["type", JSON.stringify(type)]];
    fieldNames.forEach(fieldName => {
      const field = NODE_FIELDS[type][fieldName];
      if (builderNames.includes(fieldName)) {
        const bindingIdentifierName = toBindingIdentifierName(fieldName);
        objectFields.push([fieldName, bindingIdentifierName]);
      } else if (!field.optional) {
        const def = JSON.stringify(field.default);
        objectFields.push([fieldName, def]);
      }
    });

    output += `${
      formatedBuilderNameLocal === formatedBuilderName ? "export " : ""
    }function ${formatedBuilderNameLocal}(${defArgs.join(", ")}): t.${type} {`;

    const nodeObjectExpression = `{\n${objectFields
      .map(([k, v]) => (k === v ? `    ${k},` : `    ${k}: ${v},`))
      .join("\n")}\n  }`;

    if (builderNames.length > 0) {
      output += `\n  return validateNode<t.${type}>(${nodeObjectExpression});`;
    } else {
      output += `\n  return ${nodeObjectExpression};`;
    }
    output += `\n}\n`;

    if (formatedBuilderNameLocal !== formatedBuilderName) {
      output += `export { ${formatedBuilderNameLocal} as ${formatedBuilderName} };\n`;
    }

    // This is needed for backwards compatibility.
    // It should be removed in the next major version.
    // JSXIdentifier -> jSXIdentifier
    if (/^[A-Z]{2}/.test(type)) {
      output += `export { ${formatedBuilderNameLocal} as ${lowerFirst(
        type
      )} }\n`;
    }
  });

  Object.keys(DEPRECATED_KEYS).forEach(type => {
    const newType = DEPRECATED_KEYS[type];
    const formatedBuilderName = formatBuilderName(type);
    const formatedNewBuilderName = formatBuilderName(newType);
    output += `/** @deprecated */
function ${type}(${generateBuilderArgs(newType).join(", ")}) {
  console.trace("The node type ${type} has been renamed to ${newType}");
  return ${formatedNewBuilderName}(${BUILDER_KEYS[newType].join(", ")});
}
export { ${type} as ${formatedBuilderName} };\n`;
    // This is needed for backwards compatibility.
    // It should be removed in the next major version.
    // JSXIdentifier -> jSXIdentifier
    if (/^[A-Z]{2}/.test(type)) {
      output += `export { ${type} as ${lowerFirst(type)} }\n`;
    }
  });

  return output;
}

function generateUppercaseBuilders() {
  let output = `/*
 * This file is auto-generated! Do not modify it directly.
 * To re-generate run 'make build'
 */

/**
 * This file is written in JavaScript and not TypeScript because uppercase builders
 * conflict with AST types. TypeScript reads the uppercase.d.ts file instead.
 */

 export {\n`;

  Object.keys(BUILDER_KEYS).forEach(type => {
    const formatedBuilderName = formatBuilderName(type);
    output += `  ${formatedBuilderName} as ${type},\n`;
  });

  Object.keys(DEPRECATED_KEYS).forEach(type => {
    const formatedBuilderName = formatBuilderName(type);
    output += `  ${formatedBuilderName} as ${type},\n`;
  });

  output += ` } from './index';\n`;
  return output;
}


text:
package com.hazelcast.internal.yaml;

/**
 * Generic YAML node interface
 */
public interface YamlNode {
    String UNNAMED_NODE = "<unnamed>";

    /**
     * Returns the parent of the given node
     *
     * @return the parent node if exists, null otherwise
     */
    YamlNode parent();

    /**
     * Returns the name of the node
     *
     * @return the name of the node or {@link #UNNAMED_NODE} if not available
     */
    String nodeName();

    /**
     * Returns the path of the node from the root of the YAML structure.
     *
     * @return the path of the node
     */
    String path();

}


text:
//
//
//package com.shapesecurity.shift.minifier;
//
//public class ReductionRule extends MinificationRule {
//}


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">

<title>Uses of Interface org.osate.xtext.aadl2.agcl.agcl.AgclPackage</title>
<meta name="date" content="2014-06-11">
<link rel="stylesheet" type="text/css" href="../../../../../../../stylesheet.css" title="Style">

JavaScript is disabled on your browser.

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev

Next

Frames

No Frames

All Classes

Uses of Interface
org.osate.xtext.aadl2.agcl.agcl.AgclPackage

Packages that use AgclPackage

Package
Description

org.osate.xtext.aadl2.agcl.agcl

org.osate.xtext.aadl2.agcl.agcl.impl

Uses of AgclPackage in org.osate.xtext.aadl2.agcl.agcl

Fields in org.osate.xtext.aadl2.agcl.agcl declared as AgclPackage

Modifier and Type
Field and Description

static AgclPackage
AgclPackage.eINSTANCE

The singleton instance of the package.

Methods in org.osate.xtext.aadl2.agcl.agcl that return AgclPackage

Modifier and Type
Method and Description

AgclPackage
AgclFactory.getAgclPackage()

Returns the package supported by this factory.

Uses of AgclPackage in org.osate.xtext.aadl2.agcl.agcl.impl

Classes in org.osate.xtext.aadl2.agcl.agcl.impl that implement AgclPackage

Modifier and Type
Class and Description

class
AgclPackageImpl

An implementation of the model Package.

Methods in org.osate.xtext.aadl2.agcl.agcl.impl that return AgclPackage

Modifier and Type
Method and Description

AgclPackage
AgclFactoryImpl.getAgclPackage()

static AgclPackage
AgclFactoryImpl.getPackage()

Deprecated.

static AgclPackage
AgclPackageImpl.init()

Creates, registers, and initializes the Package for this model, and for any others upon which it depends.

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev

Next

Frames

No Frames

All Classes


text:
<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">

<title>JRDesignReportTemplate (JasperReports 5.6.0 API)</title>
<meta name="date" content="2014-05-27">
<link rel="stylesheet" type="text/css" href="../../../../../stylesheet.css" title="Style">

JavaScript is disabled on your browser.

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev Class

Next Class

Frames

No Frames

All Classes

Summary:

Nested |

Field |

Constr |

Method

Detail:

Field |

Constr |

Method

net.sf.jasperreports.engine.design

Class JRDesignReportTemplate

java.lang.Object

net.sf.jasperreports.engine.base.JRBaseReportTemplate

net.sf.jasperreports.engine.design.JRDesignReportTemplate

All Implemented Interfaces:

java.io.Serializable, java.lang.Cloneable, JRChangeEventsSupport, JRCloneable, JRReportTemplate

public class JRDesignReportTemplate
extends JRBaseReportTemplate
implements JRChangeEventsSupport

JRReportTemplate implementation to be used at report design time.

Version:

$Id: JRDesignReportTemplate.java 5878 2013-01-07 20:23:13Z teodord $

Author:

Lucian Chirita (lucianc@users.sourceforge.net)

See Also:

Serialized Form

Field Summary

Fields

Modifier and Type
Field and Description

static java.lang.String
PROPERTY_SOURCE_EXPRESSION

Fields inherited from class net.sf.jasperreports.engine.base.JRBaseReportTemplate

sourceExpression

Constructor Summary

Constructors

Constructor and Description

JRDesignReportTemplate()

Creates an empty report template.

JRDesignReportTemplate(JRExpression sourceExpression)

Creates a report template for a template source expression.

Method Summary

Methods

Modifier and Type
Method and Description

java.lang.Object
clone()

JRPropertyChangeSupport
getEventSupport()

Returns the property change support object for this instance.

void
setSourceExpression(JRExpression sourceExpression)

Sets the template source expression.

Methods inherited from class net.sf.jasperreports.engine.base.JRBaseReportTemplate

getSourceExpression

Methods inherited from class java.lang.Object

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

Field Detail

PROPERTY_SOURCE_EXPRESSION

public static final java.lang.String PROPERTY_SOURCE_EXPRESSION

See Also:

Constant Field Values

Constructor Detail

JRDesignReportTemplate

public JRDesignReportTemplate()

Creates an empty report template.

JRDesignReportTemplate

public JRDesignReportTemplate(JRExpression sourceExpression)

Creates a report template for a template source expression.

Method Detail

setSourceExpression

public void setSourceExpression(JRExpression sourceExpression)

Sets the template source expression.

Parameters:

sourceExpression - the template source expression

See Also:

JRBaseReportTemplate.getSourceExpression()

clone

public java.lang.Object clone()

Specified by:

clone in interface JRCloneable

Overrides:

clone in class JRBaseReportTemplate

getEventSupport

public JRPropertyChangeSupport getEventSupport()

Description copied from interface: JRChangeEventsSupport

Returns the property change support object for this instance.

Specified by:

getEventSupport in interface JRChangeEventsSupport

Returns:

the property change support object for this instance

Overview

Package

Class

Use

Tree

Deprecated

Index

Help

Prev Class

Next Class

Frames

No Frames

All Classes

Summary:

Nested |

Field |

Constr |

Method

Detail:

Field |

Constr |

Method

© 2001-2010 Jaspersoft Corporation www.jaspersoft.com


text:
<link rel="stylesheet" type="text/css" href="css/changepic.css" />
<link rel="stylesheet" type="text/css" href="/shared/vendor/js/jquery-plugins/jcrop/jquery.Jcrop.css" />

                <button type="button" class="close" data-dismiss="modal" aria-label="__MSG__CLOSE__">
                    ×
                </button>

__MSG__PROFILE_PICTURE__

            <form id="changepic-form" role="form">

                                0%

                            __MSG__UPLOADING__

                            __MSG__CROPPING__

                    <button type="button" class="btn btn-link" data-dismiss="modal">__MSG__CANCEL__</button>
                    <button type="submit" id="changepic-set" class="btn btn-primary">__MSG__SET_PICTURE__</button>

            </form>


text:
using System;
//obavezno navodimo prostor imena System.Collections kako bi smo
//mogli da koristimo klase kolekcija
using System.Collections;

namespace vjezba01
{
    class Program
    {
        static void Main(string[] args)
        {
            //deklarisemo deklaraciju tipa ArrayList
            ArrayList cities = new ArrayList();

            //dodajemo svaki grad posebno
            cities.Add("Novi Sad");
            cities.Add("Beograd");
            cities.Add("Nis");

            //brisemo grad nis
            cities.Remove("Nis");

            //iscitava sva imena
            foreach (string city in cities)
                Console.WriteLine(city);

            Console.Read();
        }
    }
}


text:
Lazed
===

Lazed is a fast-paced 2D action game made using the [D](http://dlang.org)
language and [Allegro](https://allegro.cc/) gamedev library.

The source code is released under the
[MIT License](http://opensource.org/licenses/MIT)

The music, sounds, and art are available under
[CC-BY-SA 3.0](http://creativecommons.org/licenses/by-sa/3.0/).

The font used is
[Mecha by Captain Falcon](http://www.fontspace.com/captain-falcon/mecha)

# Credits:

- Written in [D](http://dlang.org)
- Using the [Allegro](https://allegro.cc/) gamedev library
- Pixel art created with [Aseprite](http://aseprite.org)
- Music created with [LMMS](https://lmms.io)
- Maps created with [Tiled](http://mapeditor.org)
- Sounds processed with [Audacity](http://www.audacityteam.org/)
- Font: [Mecha by Captain Falcon](http://www.fontspace.com/captain-falcon/mecha)
- Some sounds were generated with [bxfr](http://www.bfxr.net/)

# Other thanks:

- [D ENet Tutorials](https://github.com/MrSmith33/enet-tutorials)


text: More charts will be added later!


text:
package org.apache.flink.table.planner.plan.utils

import org.apache.flink.table.api.TableConfig
import org.apache.flink.table.planner.calcite.FlinkTypeFactory
import org.apache.flink.table.planner.codegen.ExpressionReducer
import org.apache.flink.table.planner.plan.nodes.calcite.Rank
import org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalRank
import org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalDeduplicate
import org.apache.flink.table.runtime.operators.rank.{ConstantRankRange, ConstantRankRangeWithoutEnd, RankRange, RankType, VariableRankRange}

import org.apache.calcite.plan.RelOptUtil
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.RelCollation
import org.apache.calcite.rex.{RexBuilder, RexCall, RexInputRef, RexLiteral, RexNode, RexUtil}
import org.apache.calcite.sql.SqlKind

import java.util

import scala.collection.JavaConversions._

/** Util for [[Rank]]s. */
object RankUtil {

  private[this] case class LimitPredicate(rankOnLeftSide: Boolean, pred: RexCall)

  sealed private[this] trait Boundary

  private[this] case class LowerBoundary(lower: Long) extends Boundary

  private[this] case class UpperBoundary(upper: Long) extends Boundary

  private[this] case class BothBoundary(lower: Long, upper: Long) extends Boundary

  private[this] case class InputRefBoundary(inputFieldIndex: Int) extends Boundary

  sealed private[this] trait BoundDefine

  private[this] object Lower extends BoundDefine // defined lower bound
  private[this] object Upper extends BoundDefine // defined upper bound
  private[this] object Both extends BoundDefine // defined lower and uppper bound

  /**
   * Extracts the TopN offset and fetch bounds from a predicate.
   *
   * @param oriPred
   *   the original predicate
   * @param rankFieldIndex
   *   the index of rank field
   * @param rexBuilder
   *   RexBuilder
   * @param tableConfig
   *   TableConfig
   * @return
   *   A Tuple2 of extracted rank range and remaining predicates.
   */
  def extractRankRange(
      oriPred: RexNode,
      rankFieldIndex: Int,
      rexBuilder: RexBuilder,
      tableConfig: TableConfig,
      classLoader: ClassLoader): (Option[RankRange], Option[RexNode]) = {
    val predicate = FlinkRexUtil.expandSearch(rexBuilder, oriPred)
    // Converts the condition to conjunctive normal form (CNF)
    val cnfNodeCount = tableConfig.get(FlinkRexUtil.TABLE_OPTIMIZER_CNF_NODES_LIMIT)
    val cnfCondition = FlinkRexUtil.toCnf(rexBuilder, cnfNodeCount, predicate)

    // split the condition into sort limit condition and other condition
    val (limitPreds: Seq[LimitPredicate], otherPreds: Seq[RexNode]) = cnfCondition match {
      case c: RexCall if c.getKind == SqlKind.AND =>
        c.getOperands
          .map(identifyLimitPredicate(_, rankFieldIndex))
          .foldLeft((Seq[LimitPredicate](), Seq[RexNode]())) {
            (preds, analyzed) =>
              analyzed match {
                case Left(limitPred) => (preds._1 :+ limitPred, preds._2)
                case Right(otherPred) => (preds._1, preds._2 :+ otherPred)
              }
          }
      case rex: RexNode =>
        identifyLimitPredicate(rex, rankFieldIndex) match {
          case Left(limitPred) => (Seq(limitPred), Seq())
          case Right(otherPred) => (Seq(), Seq(otherPred))
        }
      case _ =>
        return (None, Some(predicate))
    }

    if (limitPreds.isEmpty) {
      // no valid TopN bounds.
      return (None, Some(predicate))
    }

    val sortBounds =
      limitPreds.map(computeWindowBoundFromPredicate(_, rexBuilder, tableConfig, classLoader))
    val rankRange = sortBounds match {
      case Seq(Some(LowerBoundary(x)), Some(UpperBoundary(y))) =>
        new ConstantRankRange(x, y)
      case Seq(Some(UpperBoundary(x)), Some(LowerBoundary(y))) =>
        new ConstantRankRange(y, x)
      case Seq(Some(LowerBoundary(x))) =>
        // only offset
        new ConstantRankRangeWithoutEnd(x)
      case Seq(Some(UpperBoundary(x))) =>
        // rankStart starts from one
        new ConstantRankRange(1, x)
      case Seq(Some(BothBoundary(x, y))) =>
        // nth rank
        new ConstantRankRange(x, y)
      case Seq(Some(InputRefBoundary(x))) =>
        new VariableRankRange(x)
      case _ =>
        // TopN requires at least one rank comparison predicate
        return (None, Some(predicate))
    }

    val remainCondition = otherPreds match {
      case Seq() => None
      case _ => Some(otherPreds.reduceLeft((l, r) => RelOptUtil.andJoinFilters(rexBuilder, l, r)))
    }

    (Some(rankRange), remainCondition)
  }

  /**
   * Analyzes a predicate and identifies whether it is a valid predicate for a TopN. A valid TopN
   * predicate is a comparison predicate (<, <=, =>, >) or equal predicate that accesses rank fields
   * of input rel node, the rank field reference must be on one side of the condition alone.
   *
   * Examples:
   *   - rank <= 10 => valid (Top 10)
   *   - rank + 1 <= 10 => invalid: rank is not alone in the condition
   *   - rank == 10 => valid (10th)
   *   - rank <= rank + 2 => invalid: rank on same side
   *
   * @return
   *   Either a valid time predicate (Left) or a valid non-time predicate (Right)
   */
  private def identifyLimitPredicate(
      pred: RexNode,
      rankFieldIndex: Int): Either[LimitPredicate, RexNode] = pred match {
    case c: RexCall =>
      c.getKind match {
        case SqlKind.GREATER_THAN | SqlKind.GREATER_THAN_OR_EQUAL | SqlKind.LESS_THAN |
            SqlKind.LESS_THAN_OR_EQUAL | SqlKind.EQUALS =>
          val leftTerm = c.getOperands.head
          val rightTerm = c.getOperands.last

          if (
            isRankFieldRef(leftTerm, rankFieldIndex) &&
            !accessesRankField(rightTerm, rankFieldIndex)
          ) {
            Left(LimitPredicate(rankOnLeftSide = true, c))
          } else if (
            isRankFieldRef(rightTerm, rankFieldIndex) &&
            !accessesRankField(leftTerm, rankFieldIndex)
          ) {
            Left(LimitPredicate(rankOnLeftSide = false, c))
          } else {
            Right(pred)
          }

        // not a comparison predicate.
        case _ => Right(pred)
      }
    case _ => Right(pred)
  }

  // checks if the expression is the rank field reference
  def isRankFieldRef(expr: RexNode, rankFieldIndex: Int): Boolean = expr match {
    case i: RexInputRef => i.getIndex == rankFieldIndex
    case _ => false
  }

  /**
   * Checks if an expression accesses a rank field.
   *
   * @param expr
   *   The expression to check.
   * @param rankFieldIndex
   *   The rank field index.
   * @return
   *   True, if the expression accesses a time attribute. False otherwise.
   */
  def accessesRankField(expr: RexNode, rankFieldIndex: Int): Boolean = expr match {
    case i: RexInputRef => i.getIndex == rankFieldIndex
    case c: RexCall => c.operands.exists(accessesRankField(_, rankFieldIndex))
    case _ => false
  }

  /**
   * Computes the absolute bound on the left operand of a comparison expression and whether the
   * bound is an upper or lower bound.
   *
   * @return
   *   sort boundary (lower boundary, upper boundary)
   */
  private def computeWindowBoundFromPredicate(
      limitPred: LimitPredicate,
      rexBuilder: RexBuilder,
      tableConfig: TableConfig,
      classLoader: ClassLoader): Option[Boundary] = {

    val bound: BoundDefine = limitPred.pred.getKind match {
      case SqlKind.GREATER_THAN | SqlKind.GREATER_THAN_OR_EQUAL if limitPred.rankOnLeftSide =>
        Lower
      case SqlKind.GREATER_THAN | SqlKind.GREATER_THAN_OR_EQUAL if !limitPred.rankOnLeftSide =>
        Upper
      case SqlKind.LESS_THAN | SqlKind.LESS_THAN_OR_EQUAL if limitPred.rankOnLeftSide =>
        Upper
      case SqlKind.LESS_THAN | SqlKind.LESS_THAN_OR_EQUAL if !limitPred.rankOnLeftSide =>
        Lower
      case SqlKind.EQUALS =>
        Both
    }

    val predExpression = if (limitPred.rankOnLeftSide) {
      limitPred.pred.operands.get(1)
    } else {
      limitPred.pred.operands.get(0)
    }

    (predExpression, bound) match {
      case (r: RexInputRef, Upper | Both) => Some(InputRefBoundary(r.getIndex))
      case (_: RexInputRef, Lower) => None
      case _ =>
        // reduce predicate to constants to compute bounds
        val literal = reduceComparisonPredicate(limitPred, rexBuilder, tableConfig, classLoader)
        if (literal.isEmpty) {
          None
        } else {
          // compute boundary
          val tmpBoundary: Long = literal.get
          val boundary = limitPred.pred.getKind match {
            case SqlKind.LESS_THAN if limitPred.rankOnLeftSide =>
              tmpBoundary - 1
            case SqlKind.LESS_THAN =>
              tmpBoundary + 1
            case SqlKind.GREATER_THAN if limitPred.rankOnLeftSide =>
              tmpBoundary + 1
            case SqlKind.GREATER_THAN =>
              tmpBoundary - 1
            case _ =>
              tmpBoundary
          }
          bound match {
            case Lower => Some(LowerBoundary(boundary))
            case Upper => Some(UpperBoundary(boundary))
            case Both => Some(BothBoundary(boundary, boundary))
          }
        }
    }
  }

  /**
   * Replaces the rank aggregate reference on of a predicate by a zero literal and reduces the
   * expressions on both sides to a long literal.
   *
   * @param limitPred
   *   The limit predicate which both sides are reduced.
   * @param rexBuilder
   *   A RexBuilder
   * @param tableConfig
   *   A TableConfig.
   * @return
   *   The values of the reduced literals on both sides of the comparison predicate.
   */
  private def reduceComparisonPredicate(
      limitPred: LimitPredicate,
      rexBuilder: RexBuilder,
      tableConfig: TableConfig,
      classLoader: ClassLoader): Option[Long] = {

    val expression = if (limitPred.rankOnLeftSide) {
      limitPred.pred.operands.get(1)
    } else {
      limitPred.pred.operands.get(0)
    }

    if (!RexUtil.isConstant(expression)) {
      return None
    }

    // reduce expression to literal
    val exprReducer = new ExpressionReducer(tableConfig, classLoader)
    val originList = new util.ArrayList[RexNode]()
    originList.add(expression)
    val reduceList = new util.ArrayList[RexNode]()
    exprReducer.reduce(rexBuilder, originList, reduceList)

    // extract bounds from reduced literal
    val literals = reduceList.map {
      case literal: RexLiteral => Some(literal.getValue2.asInstanceOf[Long])
      case _ => None
    }

    literals.head
  }

  def isTop1(rankRange: RankRange): Boolean = rankRange match {
    case crg: ConstantRankRange => crg.getRankStart == 1L && crg.getRankEnd == 1L
    case _ => false
  }

  def getRankNumberColumnIndex(rank: Rank): Option[Int] = {
    if (rank.outputRankNumber) {
      require(rank.getRowType.getFieldCount == rank.getInput.getRowType.getFieldCount + 1)
      Some(rank.getRowType.getFieldCount - 1)
    } else {
      require(rank.getRowType.getFieldCount == rank.getInput.getRowType.getFieldCount)
      None
    }
  }

  /**
   * Whether the given rank could be converted to [[StreamPhysicalDeduplicate]].
   *
   * Returns true if the given rank is sorted by time attribute and limits 1 and its RankFunction is
   * ROW_NUMBER, else false.
   *
   * @param rank
   *   The [[FlinkLogicalRank]] node
   * @return
   *   True if the input rank could be converted to [[StreamPhysicalDeduplicate]]
   */
  def canConvertToDeduplicate(rank: FlinkLogicalRank): Boolean = {
    val sortCollation = rank.orderKey
    val rankRange = rank.rankRange

    val isRowNumberType = rank.rankType == RankType.ROW_NUMBER

    val isLimit1 = rankRange match {
      case rankRange: ConstantRankRange =>
        rankRange.getRankStart == 1 && rankRange.getRankEnd == 1
      case _ => false
    }

    val inputRowType = rank.getInput.getRowType
    val isSortOnTimeAttribute = sortOnTimeAttribute(sortCollation, inputRowType)

    !rank.outputRankNumber && isLimit1 && isSortOnTimeAttribute && isRowNumberType
  }

  private def sortOnTimeAttribute(
      sortCollation: RelCollation,
      inputRowType: RelDataType): Boolean = {
    if (sortCollation.getFieldCollations.size() != 1) {
      false
    } else {
      val firstSortField = sortCollation.getFieldCollations.get(0)
      val fieldType = inputRowType.getFieldList.get(firstSortField.getFieldIndex).getType
      FlinkTypeFactory.isProctimeIndicatorType(fieldType) ||
      FlinkTypeFactory.isRowtimeIndicatorType(fieldType)
    }
  }
}


text:
package mil.nga.giat.mage.sdk.datastore.observation;

import android.content.Context;
import android.util.Log;

import com.j256.ormlite.dao.Dao;
import com.j256.ormlite.stmt.QueryBuilder;

import java.security.SecureRandom;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import mil.nga.giat.mage.sdk.datastore.DaoHelper;
import mil.nga.giat.mage.sdk.event.IAttachmentEventListener;
import mil.nga.giat.mage.sdk.event.IEventDispatcher;
import mil.nga.giat.mage.sdk.exceptions.ObservationException;

public class AttachmentHelper extends DaoHelper<Attachment> implements IEventDispatcher<IAttachmentEventListener> {

	private static final String LOG_NAME = AttachmentHelper.class.getName();

	private static final SecureRandom random = new SecureRandom();

	private final Dao<Attachment, Long> attachmentDao;

	private Collection<IAttachmentEventListener> listeners = new CopyOnWriteArrayList<IAttachmentEventListener>();

	/**
	 * Singleton.
	 */
	private static AttachmentHelper attachmentHelper;

	/**
	 * Use of a Singleton here ensures that an excessive amount of DAOs are not
	 * created.
	 *
	 * @param context
	 *            Application Context
	 * @return A fully constructed and operational ObservationHelper.
	 */
	public static AttachmentHelper getInstance(Context context) {
		if (attachmentHelper == null) {
			attachmentHelper = new AttachmentHelper(context);
		}
		return attachmentHelper;
	}

	/**
	 * Only one-per JVM. Singleton.
	 *
	 * @param context
	 */
	private AttachmentHelper(Context context) {
		super(context);

		try {
			// Set up DAOs
			attachmentDao = daoStore.getAttachmentDao();
		} catch (SQLException sqle) {
			Log.e(LOG_NAME, "Unable to communicate with Attachment database.", sqle);

			throw new IllegalStateException("Unable to communicate with Attachment database.", sqle);
		}
	}

	@Override
	public Attachment read(Long id) throws Exception {
		Attachment attachment;
		try {
			attachment = attachmentDao.queryForId(id);
		} catch (SQLException sqle) {
			Log.e(LOG_NAME, "Unable to read Attachment: " + id, sqle);
			throw new ObservationException("Unable to read Attachment: " + id, sqle);
		}

		return attachment;
	}

	@Override
	public Attachment read(String remoteId) throws Exception {
		Attachment attachment = null;
		try {
			List<Attachment> results = attachmentDao.queryBuilder().where().eq("remote_id", remoteId).query();
			if (results != null && results.size() > 0) {
				attachment = results.get(0);
			}
		} catch (SQLException sqle) {
			Log.e(LOG_NAME, "Unable to read Attachment: " + remoteId, sqle);
			throw new ObservationException("Unable to read Attachment: " + remoteId, sqle);
		}

		return attachment;
	}

	@Override
	public Attachment create(Attachment attachment) throws SQLException {
		try {
			Attachment oldAttachment = read(attachment.getId());
			if (oldAttachment != null && attachment.getLocalPath() == null) {
				attachment.setLocalPath(oldAttachment.getLocalPath());
			}
		} catch (Exception e) {
		}

		attachmentDao.createOrUpdate(attachment);

		for (IAttachmentEventListener listener : listeners) {
			listener.onAttachmentCreated(attachment);
		}

		return attachment;
	}

	/**
	 *  Persist attachment to database.
	 *
	 * The localPath member will not be set to null (removed).  Please use
	 * the  {@link AttachmentHelper#removeLocalPath(Attachment)} method to
	 * set the localPath to null.
	 *
	 * @param attachment
	 * @return the attachment
	 * @throws SQLException
	 */
	@Override
	public Attachment update(Attachment attachment) throws SQLException {
		try {
			Attachment oldAttachment = read(attachment.getId());
			if (oldAttachment != null && attachment.getLocalPath() == null) {
				attachment.setLocalPath(oldAttachment.getLocalPath());
			}
		} catch (Exception e) {
		}

		attachmentDao.update(attachment);

		for (IAttachmentEventListener listener : listeners) {
			listener.onAttachmentUpdated(attachment);
		}

		return attachment;
	}

	/**
	 * Removes this attachments local path.
	 *
	 * The localPath member cannot be set to null (removed) from the update method.
	 * This is to protect from overriding a local path value when we pull updates for this
	 * attachment from the server since localPath does not come from the server and
	 * will always be null.
	 *
	 * @param attachment
	 * @return the attachment
	 * @throws SQLException
	 */
	public Attachment removeLocalPath(Attachment attachment) throws SQLException {
		attachment.setLocalPath(null);
		attachmentDao.update(attachment);

		return attachment;
	}

	/**
	 * Deletes an Attachment.
	 *
	 * @param attachment
	 * @throws Exception
	 */
	public void delete(Attachment attachment) throws SQLException {
		attachmentDao.deleteById(attachment.getId());

		for (IAttachmentEventListener listener : listeners) {
			listener.onAttachmentDeleted(attachment);
		}
	}

	/**
	 * A List of {@link Attachment} from the datastore that are dirty (i.e.
	 * should be synced with the server).
	 *
	 * @return
	 */
	public List<Attachment> getDirtyAttachments() {
		QueryBuilder<Attachment, Long> qb = attachmentDao.queryBuilder();

		QueryBuilder<Attachment, Long> queryBuilder = attachmentDao.queryBuilder();
		List<Attachment> attachments = new ArrayList<Attachment>();

		try {
			long count = qb.countOf();

			queryBuilder.where().eq("dirty", true);
			attachments = attachmentDao.query(queryBuilder.prepare());
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			Log.e(LOG_NAME, "Could not get dirty Attachments.");
		}
		return attachments;
	}

	@Override
	public boolean addListener(IAttachmentEventListener listener) {
		return listeners.add(listener);
	}

	@Override
	public boolean removeListener(IAttachmentEventListener listener) {
		return listeners.remove(listener);
	}

	public void uploadableAttachment(Attachment attachment) {
		for (IAttachmentEventListener listener : listeners) {
			listener.onAttachmentUploadable(attachment);
		}
	}
}