text:
package org.apache.flink.table.runtime.operators.window;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.checkpoint.OperatorSubtaskState;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.util.KeyedOneInputStreamOperatorTestHarness;
import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.JoinedRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.util.RowDataUtil;
import org.apache.flink.table.runtime.dataview.StateDataViewStore;
import org.apache.flink.table.runtime.generated.NamespaceAggsHandleFunction;
import org.apache.flink.table.runtime.generated.NamespaceAggsHandleFunctionBase;
import org.apache.flink.table.runtime.generated.NamespaceTableAggsHandleFunction;
import org.apache.flink.table.runtime.generated.RecordEqualiser;
import org.apache.flink.table.runtime.operators.window.assigners.SessionWindowAssigner;
import org.apache.flink.table.runtime.operators.window.assigners.TumblingWindowAssigner;
import org.apache.flink.table.runtime.operators.window.assigners.WindowAssigner;
import org.apache.flink.table.runtime.operators.window.triggers.ElementTriggers;
import org.apache.flink.table.runtime.operators.window.triggers.EventTimeTriggers;
import org.apache.flink.table.runtime.operators.window.triggers.ProcessingTimeTriggers;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.runtime.util.BinaryRowDataKeySelector;
import org.apache.flink.table.runtime.util.GenericRowRecordSortComparator;
import org.apache.flink.table.runtime.util.RowDataHarnessAssertor;
import org.apache.flink.table.types.logical.BigIntType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.util.Collector;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

import static org.apache.flink.table.data.StringData.fromString;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateAfterRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateBeforeRecord;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

/**
 * {@link WindowOperator} tests for {@link AggregateWindowOperator} or
 * {@link TableAggregateWindowOperator}.
 *
 * <p>To simplify the testing logic, the table aggregate outputs same value with the aggregate
 * except that the table aggregate outputs two same records each time.
 */
@RunWith(Parameterized.class)
public class WindowOperatorTest {

	@Parameterized.Parameters(name = "isTableAggregate = {0}")
	public static Collection<Object[]> runMode() {
		return Arrays.asList(
			new Object[] { false },
			new Object[] { true });
	}

	private final boolean isTableAggregate;
	private static final SumAndCountAggTimeWindow sumAndCountAggTimeWindow =
		new SumAndCountAggTimeWindow();
	private static final SumAndCountTableAggTimeWindow sumAndCountTableAggTimeWindow =
		new SumAndCountTableAggTimeWindow();
	private static final SumAndCountAggCountWindow sumAndCountAggCountWindow =
		new SumAndCountAggCountWindow();
	private static final SumAndCountTableAggCountWindow sumAndCountTableAggCountWindow =
		new SumAndCountTableAggCountWindow();

	public WindowOperatorTest(boolean isTableAggregate) {
		this.isTableAggregate = isTableAggregate;
	}

	private NamespaceAggsHandleFunctionBase getTimeWindowAggFunction() {
		return isTableAggregate ? sumAndCountTableAggTimeWindow : sumAndCountAggTimeWindow;
	}

	private NamespaceAggsHandleFunctionBase getCountWindowAggFunction() {
		return isTableAggregate ? sumAndCountTableAggCountWindow : sumAndCountAggCountWindow;
	}

	// For counting if close() is called the correct number of times on the SumReducer
	private static AtomicInteger closeCalled = new AtomicInteger(0);

	private LogicalType[] inputFieldTypes = new LogicalType[]{
			new VarCharType(VarCharType.MAX_LENGTH),
			new IntType(),
			new BigIntType()};

	private InternalTypeInfo<RowData> outputType = InternalTypeInfo.ofFields(
			new VarCharType(VarCharType.MAX_LENGTH),
			new BigIntType(),
			new BigIntType(),
			new BigIntType(),
			new BigIntType(),
			new BigIntType());

	private LogicalType[] aggResultTypes = new LogicalType[] { new BigIntType(), new BigIntType() };
	private LogicalType[] accTypes = new LogicalType[] { new BigIntType(), new BigIntType() };
	private LogicalType[] windowTypes = new LogicalType[] { new BigIntType(), new BigIntType(), new BigIntType() };
	private GenericRowEqualiser equaliser = new GenericRowEqualiser(accTypes, windowTypes);
	private BinaryRowDataKeySelector keySelector = new BinaryRowDataKeySelector(new int[] { 0 }, inputFieldTypes);
	private TypeInformation<RowData> keyType = keySelector.getProducedType();
	private RowDataHarnessAssertor assertor = new RowDataHarnessAssertor(
			outputType.toRowFieldTypes(),
			new GenericRowRecordSortComparator(0, new VarCharType(VarCharType.MAX_LENGTH)));

	private ConcurrentLinkedQueue<Object> doubleRecord(boolean isDouble, StreamRecord<RowData> record) {
		ConcurrentLinkedQueue<Object> results = new ConcurrentLinkedQueue<>();
		results.add(record);
		if (isDouble) {
			results.add(record);
		}
		return results;
	}

	@Test
	public void testEventTimeSlidingWindows() throws Exception {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.sliding(Duration.ofSeconds(3), Duration.ofSeconds(1))
				.withEventTime(2)
				.aggregateAndBuild(getTimeWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		testHarness.open();

		// process elements
		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		// add elements out-of-order
		testHarness.processElement(insertRecord("key2", 1, 3999L));
		testHarness.processElement(insertRecord("key2", 1, 3000L));

		testHarness.processElement(insertRecord("key1", 1, 20L));
		testHarness.processElement(insertRecord("key1", 1, 0L));
		testHarness.processElement(insertRecord("key1", 1, 999L));

		testHarness.processElement(insertRecord("key2", 1, 1998L));
		testHarness.processElement(insertRecord("key2", 1, 1999L));
		testHarness.processElement(insertRecord("key2", 1, 1000L));

		testHarness.processWatermark(new Watermark(999));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 3L, 3L, -2000L, 1000L, 999L)));
		expectedOutput.add(new Watermark(999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(1999));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 3L, 3L, -1000L, 2000L, 1999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 3L, 3L, -1000L, 2000L, 1999L)));
		expectedOutput.add(new Watermark(1999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(2999));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 3L, 3L, 0L, 3000L, 2999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 3L, 3L, 0L, 3000L, 2999L)));
		expectedOutput.add(new Watermark(2999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// do a snapshot, close and restore again
		OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
		testHarness.close();
		expectedOutput.clear();

		testHarness = createTestHarness(operator);
		testHarness.setup();
		testHarness.initializeState(snapshot);
		testHarness.open();

		testHarness.processWatermark(new Watermark(3999));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 5L, 5L, 1000L, 4000L, 3999L)));
		expectedOutput.add(new Watermark(3999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(4999));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 2L, 2L, 2000L, 5000L, 4999L)));
		expectedOutput.add(new Watermark(4999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(5999));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 2L, 2L, 3000L, 6000L, 5999L)));
		expectedOutput.add(new Watermark(5999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// those don't have any effect...
		testHarness.processWatermark(new Watermark(6999));
		testHarness.processWatermark(new Watermark(7999));
		expectedOutput.add(new Watermark(6999));
		expectedOutput.add(new Watermark(7999));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();

		// we close once in the rest...
		assertEquals("Close was not called.", 2, closeCalled.get());
	}

	@Test
	public void testProcessingTimeSlidingWindows() throws Throwable {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.sliding(Duration.ofSeconds(3), Duration.ofSeconds(1))
				.withProcessingTime()
				.aggregateAndBuild(getTimeWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		// timestamp is ignored in processing time
		testHarness.setProcessingTime(3);
		testHarness.processElement(insertRecord("key2", 1, Long.MAX_VALUE));

		testHarness.setProcessingTime(1000);

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 1L, 1L, -2000L, 1000L, 999L)));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key2", 1, Long.MAX_VALUE));
		testHarness.processElement(insertRecord("key2", 1, Long.MAX_VALUE));

		testHarness.setProcessingTime(2000);

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 3L, 3L, -1000L, 2000L, 1999L)));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key1", 1, Long.MAX_VALUE));
		testHarness.processElement(insertRecord("key1", 1, Long.MAX_VALUE));

		testHarness.setProcessingTime(3000);

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 3L, 3L, 0L, 3000L, 2999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 2L, 2L, 0L, 3000L, 2999L)));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key1", 1, Long.MAX_VALUE));
		testHarness.processElement(insertRecord("key1", 1, Long.MAX_VALUE));
		testHarness.processElement(insertRecord("key1", 1, Long.MAX_VALUE));

		testHarness.setProcessingTime(7000);

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 2L, 2L, 1000L, 4000L, 3999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 5L, 5L, 1000L, 4000L, 3999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 5L, 5L, 2000L, 5000L, 4999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 3L, 3L, 3000L, 6000L, 5999L)));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();
	}

	@Test
	@SuppressWarnings("unchecked")
	public void testEventTimeTumblingWindows() throws Exception {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.tumble(Duration.ofSeconds(3))
				.withEventTime(2)
				.aggregateAndBuild(getTimeWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		// add elements out-of-order
		testHarness.processElement(insertRecord("key2", 1, 3999L));
		testHarness.processElement(insertRecord("key2", 1, 3000L));

		testHarness.processElement(insertRecord("key1", 1, 20L));
		testHarness.processElement(insertRecord("key1", 1, 0L));
		testHarness.processElement(insertRecord("key1", 1, 999L));

		testHarness.processElement(insertRecord("key2", 1, 1998L));
		testHarness.processElement(insertRecord("key2", 1, 1999L));
		testHarness.processElement(insertRecord("key2", 1, 1000L));

		testHarness.processWatermark(new Watermark(999));
		expectedOutput.add(new Watermark(999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(1999));
		expectedOutput.add(new Watermark(1999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// do a snapshot, close and restore again
		OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
		testHarness.close();
		expectedOutput.clear();

		testHarness = createTestHarness(operator);
		testHarness.setup();
		testHarness.initializeState(snapshot);
		testHarness.open();

		testHarness.processWatermark(new Watermark(2999));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 3L, 3L, 0L, 3000L, 2999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 3L, 3L, 0L, 3000L, 2999L)));
		expectedOutput.add(new Watermark(2999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(3999));
		expectedOutput.add(new Watermark(3999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(4999));
		expectedOutput.add(new Watermark(4999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(5999));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 2L, 2L, 3000L, 6000L, 5999L)));
		expectedOutput.add(new Watermark(5999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// those don't have any effect...
		testHarness.processWatermark(new Watermark(6999));
		testHarness.processWatermark(new Watermark(7999));
		expectedOutput.add(new Watermark(6999));
		expectedOutput.add(new Watermark(7999));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();

		// we close once in the rest...
		assertEquals("Close was not called.", 2, closeCalled.get());
	}

	@Test
	@SuppressWarnings("unchecked")
	public void testEventTimeTumblingWindowsWithEarlyFiring() throws Exception {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.tumble(Duration.ofSeconds(3))
				.withEventTime(2)
				.triggering(
						EventTimeTriggers
								.afterEndOfWindow()
								.withEarlyFirings(ProcessingTimeTriggers.every(Duration.ofSeconds(1))))
				.produceUpdates()
				.aggregate(new SumAndCountAggTimeWindow(), equaliser, accTypes, aggResultTypes, windowTypes)
				.build();

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();
		testHarness.setProcessingTime(0L);

		// add elements out-of-order
		testHarness.processElement(insertRecord("key2", 1, 3999L));
		testHarness.processElement(insertRecord("key2", 1, 3000L));

		testHarness.setProcessingTime(1L);
		testHarness.processElement(insertRecord("key1", 1, 20L));
		testHarness.processElement(insertRecord("key1", 1, 0L));
		testHarness.processElement(insertRecord("key1", 1, 999L));

		testHarness.processElement(insertRecord("key2", 1, 1998L));
		testHarness.processElement(insertRecord("key2", 1, 1999L));
		testHarness.processElement(insertRecord("key2", 1, 1000L));

		testHarness.setProcessingTime(1000);
		expectedOutput.add(insertRecord("key2", 2L, 2L, 3000L, 6000L, 5999L));
		testHarness.processWatermark(new Watermark(999));
		expectedOutput.add(new Watermark(999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.setProcessingTime(1001);
		expectedOutput.add(insertRecord("key1", 3L, 3L, 0L, 3000L, 2999L));
		expectedOutput.add(insertRecord("key2", 3L, 3L, 0L, 3000L, 2999L));

		testHarness.processWatermark(new Watermark(1999));
		testHarness.setProcessingTime(2001);
		expectedOutput.add(new Watermark(1999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// do a snapshot, close and restore again
		OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
		testHarness.close();
		expectedOutput.clear();

		// new a testHarness
		testHarness = createTestHarness(operator);
		testHarness.setup();
		testHarness.initializeState(snapshot);
		testHarness.open();

		testHarness.setProcessingTime(3001);
		testHarness.processWatermark(new Watermark(2999));
		// on time fire key1 & key2 [0 ~ 3000) window, but because of early firing, on time result is ignored
		expectedOutput.add(new Watermark(2999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key2", 1, 4999L));
		testHarness.processWatermark(new Watermark(3999));
		testHarness.setProcessingTime(4001);
		expectedOutput.add(new Watermark(3999));
		expectedOutput.add(updateBeforeRecord("key2", 2L, 2L, 3000L, 6000L, 5999L));
		expectedOutput.add(updateAfterRecord("key2", 3L, 3L, 3000L, 6000L, 5999L));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// late arrival
		testHarness.processElement(insertRecord("key2", 1, 2001L));
		testHarness.processElement(insertRecord("key1", 1, 2030L));
		// drop late elements
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.setProcessingTime(5100);
		testHarness.processElement(insertRecord("key2", 1, 5122L));
		testHarness.processWatermark(new Watermark(4999));
		expectedOutput.add(new Watermark(4999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(5999));
		expectedOutput.add(updateBeforeRecord("key2", 3L, 3L, 3000L, 6000L, 5999L));
		expectedOutput.add(updateAfterRecord("key2", 4L, 4L, 3000L, 6000L, 5999L));
		expectedOutput.add(new Watermark(5999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.setProcessingTime(6001);
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// those don't have any effect...
		testHarness.processWatermark(new Watermark(6999));
		testHarness.processWatermark(new Watermark(7999));
		expectedOutput.add(new Watermark(6999));
		expectedOutput.add(new Watermark(7999));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// late arrival, drop
		testHarness.processElement(insertRecord("key2", 1, 2877L));
		testHarness.processElement(insertRecord("key1", 1, 2899L));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();

		// we close once in the rest...
		assertEquals("Close was not called.", 2, closeCalled.get());
	}

	@Test
	@SuppressWarnings("unchecked")
	public void testEventTimeTumblingWindowsWithEarlyAndLateFirings() throws Exception {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.tumble(Duration.ofSeconds(3))
				.withEventTime(2)
				.triggering(
						EventTimeTriggers
								.afterEndOfWindow()
								.withEarlyFirings(ProcessingTimeTriggers.every(Duration.ofSeconds(1)))
								.withLateFirings(ElementTriggers.every()))
				.withAllowedLateness(Duration.ofSeconds(3))
				.produceUpdates()
				.aggregate(new SumAndCountAggTimeWindow(), equaliser, accTypes, aggResultTypes, windowTypes)
				.build();

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();
		testHarness.setProcessingTime(0L);

		// add elements out-of-order
		testHarness.processElement(insertRecord("key2", 1, 3999L));
		testHarness.processElement(insertRecord("key2", 1, 3000L));

		testHarness.setProcessingTime(1L);
		testHarness.processElement(insertRecord("key1", 1, 20L));
		testHarness.processElement(insertRecord("key1", 1, 0L));
		testHarness.processElement(insertRecord("key1", 1, 999L));

		testHarness.processElement(insertRecord("key2", 1, 1998L));
		testHarness.processElement(insertRecord("key2", 1, 1999L));
		testHarness.processElement(insertRecord("key2", 1, 1000L));

		testHarness.setProcessingTime(1000);
		expectedOutput.add(insertRecord("key2", 2L, 2L, 3000L, 6000L, 5999L));
		testHarness.processWatermark(new Watermark(999));
		expectedOutput.add(new Watermark(999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.setProcessingTime(1001);
		expectedOutput.add(insertRecord("key1", 3L, 3L, 0L, 3000L, 2999L));
		expectedOutput.add(insertRecord("key2", 3L, 3L, 0L, 3000L, 2999L));

		testHarness.processWatermark(new Watermark(1999));
		testHarness.setProcessingTime(2001);
		expectedOutput.add(new Watermark(1999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// do a snapshot, close and restore again
		OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
		testHarness.close();
		expectedOutput.clear();

		// new a testHarness
		testHarness = createTestHarness(operator);
		testHarness.setup();
		testHarness.initializeState(snapshot);
		testHarness.open();

		testHarness.setProcessingTime(3001);
		testHarness.processWatermark(new Watermark(2999));
		// on time fire key1 & key2 [0 ~ 3000) window, but because of early firing, on time result is ignored
		expectedOutput.add(new Watermark(2999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key2", 1, 4999L));
		testHarness.processWatermark(new Watermark(3999));
		testHarness.setProcessingTime(4001);
		expectedOutput.add(new Watermark(3999));
		expectedOutput.add(updateBeforeRecord("key2", 2L, 2L, 3000L, 6000L, 5999L));
		expectedOutput.add(updateAfterRecord("key2", 3L, 3L, 3000L, 6000L, 5999L));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// late arrival
		testHarness.processElement(insertRecord("key2", 1, 2001L));
		expectedOutput.add(updateBeforeRecord("key2", 3L, 3L, 0L, 3000L, 2999L));
		expectedOutput.add(updateAfterRecord("key2", 4L, 4L, 0L, 3000L, 2999L));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// late arrival
		testHarness.processElement(insertRecord("key1", 1, 2030L));
		expectedOutput.add(updateBeforeRecord("key1", 3L, 3L, 0L, 3000L, 2999L));
		expectedOutput.add(updateAfterRecord("key1", 4L, 4L, 0L, 3000L, 2999L));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.setProcessingTime(5100);
		testHarness.processElement(insertRecord("key2", 1, 5122L));
		testHarness.processWatermark(new Watermark(4999));
		expectedOutput.add(new Watermark(4999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processWatermark(new Watermark(5999));
		expectedOutput.add(updateBeforeRecord("key2", 3L, 3L, 3000L, 6000L, 5999L));
		expectedOutput.add(updateAfterRecord("key2", 4L, 4L, 3000L, 6000L, 5999L));
		expectedOutput.add(new Watermark(5999));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.setProcessingTime(6001);
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// those don't have any effect...
		testHarness.processWatermark(new Watermark(6999));
		testHarness.processWatermark(new Watermark(7999));
		expectedOutput.add(new Watermark(6999));
		expectedOutput.add(new Watermark(7999));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// late arrival, but too late, drop
		testHarness.processElement(insertRecord("key2", 1, 2877L));
		testHarness.processElement(insertRecord("key1", 1, 2899L));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();

		// we close once in the rest...
		assertEquals("Close was not called.", 2, closeCalled.get());
	}

	@Test
	@SuppressWarnings("unchecked")
	public void testProcessingTimeTumblingWindows() throws Exception {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder.builder()
				.withInputFields(inputFieldTypes)
				.tumble(Duration.ofSeconds(3))
				.withProcessingTime()
				.aggregateAndBuild(getTimeWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		testHarness.setProcessingTime(3);

		// timestamp is ignored in processing time
		testHarness.processElement(insertRecord("key2", 1, Long.MAX_VALUE));
		testHarness.processElement(insertRecord("key2", 1, 7000L));
		testHarness.processElement(insertRecord("key2", 1, 7000L));

		testHarness.processElement(insertRecord("key1", 1, 7000L));
		testHarness.processElement(insertRecord("key1", 1, 7000L));

		testHarness.setProcessingTime(5000);

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 3L, 3L, 0L, 3000L, 2999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 2L, 2L, 0L, 3000L, 2999L)));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key1", 1, 7000L));
		testHarness.processElement(insertRecord("key1", 1, 7000L));
		testHarness.processElement(insertRecord("key1", 1, 7000L));

		testHarness.setProcessingTime(7000);

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 3L, 3L, 3000L, 6000L, 5999L)));

		assertEquals(0L, operator.getWatermarkLatency().getValue());
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();
	}

	@Test
	@SuppressWarnings("unchecked")
	public void testEventTimeSessionWindows() throws Exception {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.session(Duration.ofSeconds(3))
				.withEventTime(2)
				.aggregateAndBuild(getTimeWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		// add elements out-of-order
		testHarness.processElement(insertRecord("key2", 1, 0L));
		testHarness.processElement(insertRecord("key2", 2, 1000L));
		testHarness.processElement(insertRecord("key2", 3, 2500L));

		testHarness.processElement(insertRecord("key1", 1, 10L));
		testHarness.processElement(insertRecord("key1", 2, 1000L));

		// do a snapshot, close and restore again
		OperatorSubtaskState snapshotV2 = testHarness.snapshot(0L, 0);
		testHarness.close();
		expectedOutput.clear();

		testHarness = createTestHarness(operator);
		testHarness.setup();
		testHarness.initializeState(snapshotV2);
		testHarness.open();

		assertEquals(0L, operator.getWatermarkLatency().getValue());

		testHarness.processElement(insertRecord("key1", 3, 2500L));

		testHarness.processElement(insertRecord("key2", 4, 5501L));
		testHarness.processElement(insertRecord("key2", 5, 6000L));
		testHarness.processElement(insertRecord("key2", 5, 6000L));
		testHarness.processElement(insertRecord("key2", 6, 6050L));

		testHarness.processWatermark(new Watermark(12000));

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 6L, 3L, 10L, 5500L, 5499L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 6L, 3L, 0L, 5500L, 5499L)));

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 20L, 4L, 5501L, 9050L, 9049L)));
		expectedOutput.add(new Watermark(12000));

		// add a late data
		testHarness.processElement(insertRecord("key1", 3, 4000L));
		testHarness.processElement(insertRecord("key2", 10, 15000L));
		testHarness.processElement(insertRecord("key2", 20, 15000L));

		testHarness.processWatermark(new Watermark(17999));

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 30L, 2L, 15000L, 18000L, 17999L)));
		expectedOutput.add(new Watermark(17999));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.setProcessingTime(18000);
		assertEquals(1L, operator.getWatermarkLatency().getValue());

		testHarness.close();

		// we close once in the rest...
		assertEquals("Close was not called.", 2, closeCalled.get());
		assertEquals(1, operator.getNumLateRecordsDropped().getCount());
	}

	@Test
	public void testProcessingTimeSessionWindows() throws Throwable {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.session(Duration.ofSeconds(3))
				.withProcessingTime()
				.aggregateAndBuild(getTimeWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		RowDataHarnessAssertor assertor = new RowDataHarnessAssertor(
			outputType.toRowFieldTypes(),
			new GenericRowRecordSortComparator(0, new VarCharType(VarCharType.MAX_LENGTH)));

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		// timestamp is ignored in processing time
		testHarness.setProcessingTime(3);
		testHarness.processElement(insertRecord("key2", 1, 1L));

		testHarness.setProcessingTime(1000);
		testHarness.processElement(insertRecord("key2", 1, 1002L));

		testHarness.setProcessingTime(5000);

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 2L, 2L, 3L, 4000L, 3999L)));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key2", 1, 5000L));
		testHarness.processElement(insertRecord("key2", 1, 5000L));
		testHarness.processElement(insertRecord("key1", 1, 5000L));
		testHarness.processElement(insertRecord("key1", 1, 5000L));
		testHarness.processElement(insertRecord("key1", 1, 5000L));

		testHarness.setProcessingTime(10000);

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 2L, 2L, 5000L, 8000L, 7999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 3L, 3L, 5000L, 8000L, 7999L)));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();
	}

	/**
	 * This tests a custom Session window assigner that assigns some elements to "point windows",
	 * windows that have the same timestamp for start and end.
	 *
	 * <p>In this test, elements that have 33 as the second tuple field will be put into a point
	 * window.
	 */
	@Test
	@SuppressWarnings("unchecked")
	public void testPointSessions() throws Exception {
		closeCalled.set(0);

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.assigner(new PointSessionWindowAssigner(3000))
				.withEventTime(2)
				.aggregateAndBuild(getTimeWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		// add elements out-of-order
		testHarness.processElement(insertRecord("key2", 1, 0L));
		testHarness.processElement(insertRecord("key2", 33, 1000L));

		// do a snapshot, close and restore again
		OperatorSubtaskState snapshot = testHarness.snapshot(0L, 0);
		testHarness.close();

		testHarness = createTestHarness(operator);
		testHarness.setup();
		testHarness.initializeState(snapshot);
		testHarness.open();

		testHarness.processElement(insertRecord("key2", 33, 2500L));

		testHarness.processElement(insertRecord("key1", 1, 10L));
		testHarness.processElement(insertRecord("key1", 2, 1000L));
		testHarness.processElement(insertRecord("key1", 33, 2500L));

		testHarness.processWatermark(new Watermark(12000));

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 36L, 3L, 10L, 4000L, 3999L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 67L, 3L, 0L, 3000L, 2999L)));
		expectedOutput.add(new Watermark(12000));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();

		// we close once in the rest...
		assertEquals("Close was not called.", 2, closeCalled.get());
	}

	@Test
	public void testLateness() throws Exception {
		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.tumble(Duration.ofSeconds(2))
				.withEventTime(2)
				.withAllowedLateness(Duration.ofMillis(500))
				.produceUpdates()
				.aggregateAndBuild(new SumAndCountAggTimeWindow(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		testHarness.processElement(insertRecord("key2", 1, 500L));
		testHarness.processWatermark(new Watermark(1500));

		expectedOutput.add(new Watermark(1500));

		testHarness.processElement(insertRecord("key2", 1, 1300L));
		testHarness.processWatermark(new Watermark(2300));

		expectedOutput.add(insertRecord("key2", 2L, 2L, 0L, 2000L, 1999L));
		expectedOutput.add(new Watermark(2300));

		// this will not be dropped because window.maxTimestamp() + allowedLateness > currentWatermark
		testHarness.processElement(insertRecord("key2", 1, 1997L));
		testHarness.processWatermark(new Watermark(6000));

		// this is 1 and not 3 because the trigger fires and purges
		expectedOutput.add(updateBeforeRecord("key2", 2L, 2L, 0L, 2000L, 1999L));
		expectedOutput.add(updateAfterRecord("key2", 3L, 3L, 0L, 2000L, 1999L));
		expectedOutput.add(new Watermark(6000));

		// this will be dropped because window.maxTimestamp() + allowedLateness < currentWatermark
		testHarness.processElement(insertRecord("key2", 1, 1998L));
		testHarness.processWatermark(new Watermark(7000));

		expectedOutput.add(new Watermark(7000));

		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		assertEquals(1, operator.getNumLateRecordsDropped().getCount());

		testHarness.close();
	}

	@Test
	public void testCleanupTimeOverflow() throws Exception {
		long windowSize = 1000;
		long lateness = 2000;
		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.tumble(Duration.ofMillis(windowSize))
				.withEventTime(2)
				.withAllowedLateness(Duration.ofMillis(lateness))
				.produceUpdates()
				.aggregateAndBuild(new SumAndCountAggTimeWindow(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
				new KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData>(
						operator, keySelector, keyType);

		testHarness.open();

		ConcurrentLinkedQueue<Object> expected = new ConcurrentLinkedQueue<>();

		WindowAssigner<TimeWindow> windowAssigner = TumblingWindowAssigner.of(Duration.ofMillis(windowSize));
		long timestamp = Long.MAX_VALUE - 1750;
		Collection<TimeWindow> windows = windowAssigner.assignWindows(GenericRowData.of(fromString("key2"), 1), timestamp);
		TimeWindow window = windows.iterator().next();

		testHarness.processElement(insertRecord("key2", 1, timestamp));

		// the garbage collection timer would wrap-around
		assertTrue(window.maxTimestamp() + lateness < window.maxTimestamp());

		// and it would prematurely fire with watermark (Long.MAX_VALUE - 1500)
		assertTrue(window.maxTimestamp() + lateness < Long.MAX_VALUE - 1500);

		// if we don't correctly prevent wrap-around in the garbage collection
		// timers this watermark will clean our window state for the just-added
		// element/window
		testHarness.processWatermark(new Watermark(Long.MAX_VALUE - 1500));

		// this watermark is before the end timestamp of our only window
		assertTrue(Long.MAX_VALUE - 1500 < window.maxTimestamp());
		assertTrue(window.maxTimestamp() < Long.MAX_VALUE);

		// push in a watermark that will trigger computation of our window
		testHarness.processWatermark(new Watermark(window.maxTimestamp()));

		expected.add(new Watermark(Long.MAX_VALUE - 1500));
		expected.add(insertRecord("key2", 1L, 1L, window.getStart(), window.getEnd(), window.maxTimestamp()));
		expected.add(new Watermark(window.maxTimestamp()));

		assertor.assertOutputEqualsSorted("Output was not correct.", expected, testHarness.getOutput());
		testHarness.close();
	}

	@Test
	public void testCleanupTimerWithEmptyReduceStateForTumblingWindows() throws Exception {
		final int windowSize = 2;
		final long lateness = 1;

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.tumble(Duration.ofSeconds(windowSize))
				.withEventTime(2)
				.withAllowedLateness(Duration.ofMillis(lateness))
				.produceUpdates()
				.aggregateAndBuild(new SumAndCountAggTimeWindow(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		testHarness.open();

		ConcurrentLinkedQueue<Object> expected = new ConcurrentLinkedQueue<>();

		// normal element
		testHarness.processElement(insertRecord("key2", 1, 1000L));
		testHarness.processWatermark(new Watermark(1599));
		testHarness.processWatermark(new Watermark(1999));
		testHarness.processWatermark(new Watermark(2000));
		testHarness.processWatermark(new Watermark(5000));

		expected.add(new Watermark(1599));
		expected.add(insertRecord("key2", 1L, 1L, 0L, 2000L, 1999L));
		expected.add(new Watermark(1999)); // here it fires and purges
		expected.add(new Watermark(2000)); // here is the cleanup timer
		expected.add(new Watermark(5000));

		assertor.assertOutputEqualsSorted("Output was not correct.", expected, testHarness.getOutput());
		testHarness.close();
	}

	@Test
	public void testTumblingCountWindow() throws Exception {
		closeCalled.set(0);
		final int windowSize = 3;
		LogicalType[] windowTypes = new LogicalType[] { new BigIntType() };

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.countWindow(windowSize)
				.aggregateAndBuild(getCountWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		testHarness.processElement(insertRecord("key2", 1, 0L));
		testHarness.processElement(insertRecord("key2", 2, 1000L));
		testHarness.processElement(insertRecord("key2", 3, 2500L));
		testHarness.processElement(insertRecord("key1", 1, 10L));
		testHarness.processElement(insertRecord("key1", 2, 1000L));

		testHarness.processWatermark(new Watermark(12000));
		testHarness.setProcessingTime(12000L);
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 6L, 3L, 0L)));
		expectedOutput.add(new Watermark(12000));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// do a snapshot, close and restore again
		OperatorSubtaskState snapshotV2 = testHarness.snapshot(0L, 0);
		testHarness.close();
		expectedOutput.clear();

		testHarness = createTestHarness(operator);
		testHarness.setup();
		testHarness.initializeState(snapshotV2);
		testHarness.open();

		testHarness.processElement(insertRecord("key1", 2, 2500L));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 5L, 3L, 0L)));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key2", 4, 5501L));
		testHarness.processElement(insertRecord("key2", 5, 6000L));
		testHarness.processElement(insertRecord("key2", 5, 6000L));
		testHarness.processElement(insertRecord("key2", 6, 6050L));

		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 14L, 3L, 1L)));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key1", 3, 4000L));
		testHarness.processElement(insertRecord("key2", 10, 15000L));
		testHarness.processElement(insertRecord("key2", 20, 15000L));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 36L, 3L, 2L)));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key1", 2, 2500L));
		testHarness.processElement(insertRecord("key1", 2, 2500L));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 7L, 3L, 1L)));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();

		// we close once in the rest...
		assertEquals("Close was not called.", 2, closeCalled.get());
	}

	@Test
	public void testSlidingCountWindow() throws Exception {
		closeCalled.set(0);
		final int windowSize = 5;
		final int windowSlide = 3;
		LogicalType[] windowTypes = new LogicalType[] { new BigIntType() };

		WindowOperator operator = WindowOperatorBuilder
				.builder()
				.withInputFields(inputFieldTypes)
				.countWindow(windowSize, windowSlide)
				.aggregateAndBuild(getCountWindowAggFunction(), equaliser, accTypes, aggResultTypes, windowTypes);

		OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(operator);

		ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

		testHarness.open();

		testHarness.processElement(insertRecord("key2", 1, 0L));
		testHarness.processElement(insertRecord("key2", 2, 1000L));
		testHarness.processElement(insertRecord("key2", 3, 2500L));
		testHarness.processElement(insertRecord("key2", 4, 2500L));
		testHarness.processElement(insertRecord("key2", 5, 2500L));
		testHarness.processElement(insertRecord("key1", 1, 10L));
		testHarness.processElement(insertRecord("key1", 2, 1000L));

		testHarness.processWatermark(new Watermark(12000));
		testHarness.setProcessingTime(12000L);
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 15L, 5L, 0L)));
		expectedOutput.add(new Watermark(12000));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		// do a snapshot, close and restore again
		OperatorSubtaskState snapshotV2 = testHarness.snapshot(0L, 0);
		testHarness.close();
		expectedOutput.clear();

		testHarness = createTestHarness(operator);
		testHarness.setup();
		testHarness.initializeState(snapshotV2);
		testHarness.open();

		testHarness.processElement(insertRecord("key1", 3, 2500L));
		testHarness.processElement(insertRecord("key1", 4, 2500L));
		testHarness.processElement(insertRecord("key1", 5, 2500L));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 15L, 5L, 0L)));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key2", 6, 6000L));
		testHarness.processElement(insertRecord("key2", 7, 6000L));
		testHarness.processElement(insertRecord("key2", 8, 6050L));
		testHarness.processElement(insertRecord("key2", 9, 6050L));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 30L, 5L, 1L)));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.processElement(insertRecord("key1", 6, 4000L));
		testHarness.processElement(insertRecord("key1", 7, 4000L));
		testHarness.processElement(insertRecord("key1", 8, 4000L));
		testHarness.processElement(insertRecord("key2", 10, 15000L));
		testHarness.processElement(insertRecord("key2", 11, 15000L));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key1", 30L, 5L, 1L)));
		expectedOutput.addAll(doubleRecord(isTableAggregate, insertRecord("key2", 45L, 5L, 2L)));
		assertor.assertOutputEqualsSorted("Output was not correct.", expectedOutput, testHarness.getOutput());

		testHarness.close();

		// we close once in the rest...
		assertEquals("Close was not called.", 2, closeCalled.get());
	}

	// --------------------------------------------------------------------------------

	private static class PointSessionWindowAssigner extends SessionWindowAssigner {
		private static final long serialVersionUID = 1L;

		private final long sessionTimeout;

		private PointSessionWindowAssigner(long sessionTimeout) {
			super(sessionTimeout, true);
			this.sessionTimeout = sessionTimeout;
		}

		private PointSessionWindowAssigner(long sessionTimeout, boolean isEventTime) {
			super(sessionTimeout, isEventTime);
			this.sessionTimeout = sessionTimeout;
		}

		@Override
		@SuppressWarnings("unchecked")
		public Collection<TimeWindow> assignWindows(RowData element, long timestamp) {
			int second = element.getInt(1);
			if (second == 33) {
				return Collections.singletonList(new TimeWindow(timestamp, timestamp));
			}
			return Collections.singletonList(new TimeWindow(timestamp, timestamp + sessionTimeout));
		}

		@Override
		public SessionWindowAssigner withEventTime() {
			return new PointSessionWindowAssigner(sessionTimeout, true);
		}

		@Override
		public SessionWindowAssigner withProcessingTime() {
			return new PointSessionWindowAssigner(sessionTimeout, false);
		}
	}

	// sum, count, window_start, window_end
	private static class SumAndCountAggTimeWindow
		extends SumAndCountAggBase<TimeWindow>
		implements NamespaceAggsHandleFunction<TimeWindow> {

		private static final long serialVersionUID = 2062031590687738047L;

		@Override
		public RowData getValue(TimeWindow namespace) throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			GenericRowData row = new GenericRowData(5);
			if (!sumIsNull) {
				row.setField(0, sum);
			}
			if (!countIsNull) {
				row.setField(1, count);
			}
			row.setField(2, namespace.getStart());
			row.setField(3, namespace.getEnd());
			row.setField(4, namespace.maxTimestamp());
			return row;
		}
	}

	// sum, count, window_id
	private static class SumAndCountAggCountWindow
		extends SumAndCountAggBase<CountWindow>
		implements NamespaceAggsHandleFunction<CountWindow> {

		private static final long serialVersionUID = -2634639678371135643L;

		@Override
		public RowData getValue(CountWindow namespace) throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			GenericRowData row = new GenericRowData(3);
			if (!sumIsNull) {
				row.setField(0, sum);
			}
			if (!countIsNull) {
				row.setField(1, count);
			}
			row.setField(2, namespace.getId());
			return row;
		}
	}

	// (table aggregate) sum, count, window_start, window_end
	private static class SumAndCountTableAggTimeWindow
		extends SumAndCountAggBase<TimeWindow>
		implements NamespaceTableAggsHandleFunction<TimeWindow>  {

		private static final long serialVersionUID = 2062031590687738047L;

		@Override
		public void emitValue(TimeWindow namespace, RowData key, Collector<RowData> out) throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			GenericRowData row = new GenericRowData(5);
			if (!sumIsNull) {
				row.setField(0, sum);
			}
			if (!countIsNull) {
				row.setField(1, count);
			}
			row.setField(2, namespace.getStart());
			row.setField(3, namespace.getEnd());
			row.setField(4, namespace.maxTimestamp());

			result.replace(key, row);
			// Simply output two lines
			out.collect(result);
			out.collect(result);
		}
	}

	// (table aggregate) sum, count, window_id
	private static class SumAndCountTableAggCountWindow
		extends SumAndCountAggBase<CountWindow>
		implements NamespaceTableAggsHandleFunction<CountWindow> {

		private static final long serialVersionUID = -2634639678371135643L;

		@Override
		public void emitValue(CountWindow namespace, RowData key, Collector<RowData> out) throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			GenericRowData row = new GenericRowData(3);
			if (!sumIsNull) {
				row.setField(0, sum);
			}
			if (!countIsNull) {
				row.setField(1, count);
			}
			row.setField(2, namespace.getId());

			result.replace(key, row);
			// Simply output two lines
			out.collect(result);
			out.collect(result);
		}
	}

	private static class SumAndCountAggBase<W extends Window> {

		boolean openCalled;

		long sum;
		boolean sumIsNull;
		long count;
		boolean countIsNull;

		protected transient JoinedRowData result;

		public void open(StateDataViewStore store) throws Exception {
			openCalled = true;
			result = new JoinedRowData();
		}

		public void setAccumulators(W namespace, RowData acc) throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			sumIsNull = acc.isNullAt(0);
			if (!sumIsNull) {
				sum = acc.getLong(0);
			}

			countIsNull = acc.isNullAt(1);
			if (!countIsNull) {
				count = acc.getLong(1);
			}
		}

		public void accumulate(RowData inputRow) throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			boolean inputIsNull = inputRow.isNullAt(1);
			if (!inputIsNull) {
				sum += inputRow.getInt(1);
				count += 1;
			}
		}

		public void retract(RowData inputRow) throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			boolean inputIsNull = inputRow.isNullAt(1);
			if (!inputIsNull) {
				sum -= inputRow.getInt(1);
				count -= 1;
			}
		}

		public void merge(W w, RowData otherAcc) throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			boolean sumIsNull2 = otherAcc.isNullAt(0);
			if (!sumIsNull2) {
				sum += otherAcc.getLong(0);
			}
			boolean countIsNull2 = otherAcc.isNullAt(1);
			if (!countIsNull2) {
				count += otherAcc.getLong(1);
			}
		}

		public RowData createAccumulators() {
			if (!openCalled) {
				fail("Open was not called");
			}
			GenericRowData acc = new GenericRowData(2);
			acc.setField(0, 0L);
			acc.setField(1, 0L);
			return acc;
		}

		public RowData getAccumulators() throws Exception {
			if (!openCalled) {
				fail("Open was not called");
			}
			GenericRowData row = new GenericRowData(2);
			if (!sumIsNull) {
				row.setField(0, sum);
			}
			if (!countIsNull) {
				row.setField(1, count);
			}
			return row;
		}

		public void cleanup(W window) {

		}

		public void close() {
			closeCalled.incrementAndGet();
		}
	}

	private static class GenericRowEqualiser implements RecordEqualiser {

		private final LogicalType[] fieldTypes;

		GenericRowEqualiser(LogicalType[] aggResultTypes, LogicalType[] windowTypes) {
			int size = aggResultTypes.length + windowTypes.length;
			this.fieldTypes = new LogicalType[size];
			for (int i = 0; i < size; i++) {
				if (i < aggResultTypes.length) {
					fieldTypes[i] = aggResultTypes[i];
				} else {
					fieldTypes[i] = windowTypes[i - aggResultTypes.length];
				}
			}
		}

		@Override
		public boolean equals(RowData row1, RowData row2) {
			GenericRowData left = RowDataUtil.toGenericRow(row1, fieldTypes);
			GenericRowData right = RowDataUtil.toGenericRow(row2, fieldTypes);
			return left.equals(right);
		}
	}

	private OneInputStreamOperatorTestHarness<RowData, RowData> createTestHarness(
			WindowOperator operator)
			throws Exception {
		return new KeyedOneInputStreamOperatorTestHarness<RowData, RowData, RowData>(
				operator, keySelector, keyType);
	}

}


text:
<RestClientRequest>
  <option name="biscuits">
    <list />
  </option>
  <option name="httpMethod" value="POST" />
  <option name="urlBase" value="http://localhost:3000" />
  <option name="urlPath" value="/movies" />
  <option name="headers">
    <list>
      <KeyValuePair>
        <option name="key" value="Accept" />
        <option name="value" value="*/*" />
      </KeyValuePair>
      <KeyValuePair>
        <option name="key" value="Cache-Control" />
        <option name="value" value="no-cache" />
      </KeyValuePair>
      <KeyValuePair>
        <option name="key" value="Content-Type" />
        <option name="value" value="application/json" />
      </KeyValuePair>
    </list>
  </option>
  <option name="parameters">
    <list />
  </option>
  <option name="parametersEnabled" value="true" />
  <option name="haveTextToSend" value="true" />
  <option name="haveFileToSend" value="false" />
  <option name="isFileUpload" value="false" />
  <option name="textToSend" value="{"title":"Die Hard With A Vengence"}" />
  <option name="filesToSend" value="" />
</RestClientRequest>

redpajama_set_name: RedPajamaGithub


text:
"""
    proxy.py
    ~~~~~~~~
    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
    Network monitoring, controls & Application development, testing, debugging.

    :copyright: (c) 2013-present by Abhinav Singh and contributors.
    :license: BSD, see LICENSE for more details.
"""
import asyncio
from typing import Any, Optional
from multiprocessing import connection

from .threadless import Threadless

class BaseRemoteExecutor(Threadless[connection.Connection]):
    """A threadless executor implementation which receives work over a connection."""

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self._loop: Optional[asyncio.AbstractEventLoop] = None

    @property
    def loop(self) -> Optional[asyncio.AbstractEventLoop]:
        if self._loop is None:
            self._loop = asyncio.get_event_loop_policy().get_event_loop()
        return self._loop

    def work_queue_fileno(self) -> Optional[int]:
        return self.work_queue.fileno()

    def close_work_queue(self) -> None:
        self.work_queue.close()

    def receive_from_work_queue(self) -> bool:
        self.work(self.work_queue.recv())
        return False


text:
<?php
/* Prototype  : proto array getimagesize(string imagefile [, array info])
 * Description: Get the size of an image as 4-element array
 * Source code: ext/standard/image.c
 * Alias to functions:
 */

/*
 * Load APP info from jpeg
 */

$arr['this'] = "will";
$arr['all'] = "be destroyed!";
$arr['APP1'] = "and this too";

getimagesize( dirname(__FILE__)."/testAPP.jpg", $arr);

foreach ($arr as $key => $value) {
	echo "$key - length: ". strlen($value) ."; md5: " . md5($value) .  "\n" ;
}

?>
===DONE===


text:
# HTTP Requests

- [HTTP Requests](#http-requests)
  - [Truy cập vào Request](#truy-cp-vao-request)
    - [Dependency Injection & Route Parameters](#dependency-injection-route-parameters)
    - [Đường dẫn Request & Phương thức](#ng-dn-request-phng-thc)
  - [PSR-7 Requests](#psr-7-requests)
  - [Input Trimming & Normalization](#input-trimming-normalization)
  - [Nhận input](#nhn-input)
    - [Nhận toàn bộ giá trị input](#nhn-toan-b-gia-tr-input)
    - [Nhận giá trị input](#nhn-gia-tr-input)
    - [Nhận dữ liệu từ câu truy vấn](#nhn-d-liu-t-cau-truy-vn)
    - [Nhận dữ liệu với Thuộc tính động](#nhn-d-liu-vi-thuc-tinh-ng)
    - [Nhận giá trị input từ JSON](#nhn-gia-tr-input-t-json)
    - [Nhận một phần của dữ liệu input](#nhn-mt-phn-ca-d-liu-input)
    - [Kiểm tra một giá trị input có tồn tại](#kim-tra-mt-gia-tr-input-co-tn-ti)
    - [Input cũ](#input-c)
    - [Cookies](#cookies)
    - [Tạo Cookie Instances](#to-cookie-instances)
  - [Files](#files)
    - [Lấy file được upload](#ly-file-c-upload)
    - [Kiểm tra upload thành công](#kim-tra-upload-thanh-cong)
    - [Đường dẫn File & Extensions](#ng-dn-file-extensions)
    - [Phương thức khác của File](#phng-thc-khac-ca-file)
    - [Chuyển vị trí file upload](#chuyn-v-tri-file-upload)
    - [Các phương thức khác với file](#cac-phng-thc-khac-vi-file)
  - [Configuring Trusted Proxies](#configuring-trusted-proxies)
    - [Trusting All Proxies](#trusting-all-proxies)

## Truy cập vào Request

Để lấy đối tượng của HTTP request hiện tại thông qua dependency injection, bạn phải type-hint `Illuminate\Http\Request` vào trong hàm khởi tạo của controller hay phương thức trong controller. Đối tượng của request hiện tại sẽ được tự động inject vào bởi [service container](container.md):

```PHP
<?php

namespace App\Http\Controllers;

use Illuminate\Http\Request;

class UserController extends Controller
{
    /**
     * Store a new user.
     *
     * @param  Request  $request
     * @return Response
     */
    public function store(Request $request)
    {
        $name = $request->input('name');

        //
    }
}
```

### Dependency Injection & Route Parameters

Nếu như hàm của controller cũng cần đầu vào từ các tham số của route, đơn giản chỉ cần ghi danh sách các đối số vào sau các dependencies. Ví dụ, nếu route được khai báo như sau:

```PHP
Route::put('user/{id}', 'UserController@update');
```

Bạn vẫn có thể type-hint `Illuminate\Http\Request` và truy cập vào tham số `id` của route bằng cách khai báo trong controller như sau:

```PHP
<?php

namespace App\Http\Controllers;

use Illuminate\Http\Request;

class UserController extends Controller
{
    /**
      * Update the specified user.
      *
      * @param  Request  $request
      * @param  string  $id
      * @return Response
      */
    public function update(Request $request, $id)
    {
        //
    }
}
```

#### Truy cập vào Request qua Route Closures

Bạn cũng có thể type-hint class ``Illuminate\Http\Request`` trong route Closure. Service sẽ tự động inject các request Closure khi nó sẽ được thực thi:

```PHP
use Illuminate\Http\Request;

Route::get('/', function (Request $request) {
    //
});
```

### Đường dẫn Request & Phương thức

Đối tượng ``Illuminate\Http\Request`` cung cập một số phương thức để kiểm tra HTTP request cho ứng dụng và kế thừa class ``Symfony\Component\HttpFoundation\Request`` . Chúng ta sẽ thảo luận một số phương thức quan trọng dưới đây.

#### Nhận đường dẫn Request

Phương thức path trả về thông tin đường dẫn của request. Vì vậy, Nếu request gửi đến là  `http://domain.com/foo/bar`, phương thức path sẽ trả về `foo/bar`:

```PHP
$uri = $request->path();
```

Phương thức ``is`` sẽ cho phép bạn xác nhận những request gửi đến có đường dẫn khớp với pattern hay không. Bạn có thể sử dụng ký tự ``*`` khi sử dụng phương thức này:

```PHP
if ($request->is('admin/*')) {
    //
}
```

#### Nhận Request URL

Để nhận đường dẫn đầy đủ URL từ request gửi đến bạn có thể sử dụng phương thức ``url`` or  ``fullUrl``. Phương thức ``url`` sẽ trả về URL không có string query, trong khi phương thức ``fullUrl`` bao gồm cả string query:

```PHP
// Without Query String...
$url = $request->url();

// With Query String...
$url = $request->fullUrl();
```

#### Nhận phương thức Request

Phương thức ``method`` sẽ trả về phương thức HTTP tương ứng với request. Bạn có thể sử dụng phương thức ``isMethod`` để xác thực phương thức HTTP khớp với string:

```PHP
$method = $request->method();

if ($request->isMethod('post')) {
    //
}
```

## PSR-7 Requests

Tiêu chuẩn của PSR-7 đặt ra các interface cho HTTP messages, bao gồm cả request và response. Nếu bạn muốn lấy một đối tượng chuẩn của PSR-7 request, bạn cần phải cài đặt một số thư viện. Laravel sử dụng component Symfony HTTP Message Bridge để convert một Laravel request và response sang mẫu tương thích với PSR-7 chuẩn:

```PHP
composer require symfony/psr-http-message-bridge

composer require zendframework/zend-diactoros
```

Khi bạn đã cài đặt các thư viện này, bạn có thể lấy một request PSR-7 bằng cách type-hint kiểu request trên route hay controller:

```PHP
use Psr\Http\Message\ServerRequestInterface;

Route::get('/', function (ServerRequestInterface $request) {
    //
});
```

>Nếu bạn trả lại một response PSR-7 từ một route hoặc controller, nó sẽ tự động convert lại thành một response của Laravel và hiển thị.

## Input Trimming & Normalization

Theo mặc định, laravel bao gồm ``TrimStrings`` và ``ConvertEmptyStringsToNull`` ở trong ngăn xếp trung gian toàn cục của ứng dụng của bạn. Những phần mềm trung gian này được liệt kê trong ngăn xếp của lớp ``App\Http\Kernel``. Các phần mềm trung gian này sẽ tự động cắt tất cả các trường chuỗi đến theo yêu cầu, cũng như chuyển đổi bất kỳ trường chuỗi rỗng nào thành ``null`` . Điều này cho phép bạn không phải lo lắng về những mối quan tâm bình thường hóa trong các tuyến đường và bộ điều khiển của bạn.

Nếu bạn muốn vô hiệu hóa hành vi này, bạn có thể xóa hai phần mềm trung gian khỏi ngăn xếp phần mềm trung gian của ứng dụng bằng cách xóa chúng khỏi thuộc tính ``$middleware``của lớp ``App\Http\Kernel`` của bạn .

## Nhận input

### Nhận toàn bộ giá trị input

Bạn có thể lấy toàn bộ giá trị input dưới dạng mảng bằng cách sử dụng phương thức ``all``:

```PHP
$input=$request->all()
```

### Nhận giá trị input

Sử dụng một vài hàm cơ bản, bạn có thể lấy input từ người dùng qua `Illuminate\Http\Request`. Bạn không cần quan tâm về các hành động của HTTP sử dụng cho request.Bất kể nó là phương thức HTTP nào, phương thức input sử dụng có thể lấy được input từ người dùng:

```PHP
$name = $request->input('name');
```

Bạn có thể truyền vào giá trị mặc định ở đối số thứ hai trong hàm `input`. Giá trị này sẽ được trả lại nếu như giá trị input không có trong request:

```PHP
$name = $request->input('name', 'Sally');
```

Khi thực hiện trên form với mảng input, bạn có thể sử dụng kí hiệu "dot" để truy cập vào mảng:

```PHP
$name = $request->input('products.0.name');

$names = $request->input('products.*.name');
```

### Nhận dữ liệu từ câu truy vấn

Khi phương thức ``input`` lấy các giá trị từ toàn bộ câu lệnh yêu cầu(bao gồm cả chuỗi truy vấn), thì phương thức ``query`` chỉ nhận giá trị từ câu truy vấn

```PHP
$name=$request->query('name')
```

Nếu như câu truy vấn yêu cầu không có giá trị thì đối số thứ 2 của phương thức sẽ được trả về:

```PHP
$name = $request->query('name', 'Helen');
```

Bạn cũng có thể gọi phương thức ``query`` mà không có đối số nào để lấy toàn bộ giá trị của câu truy vấn dưới dạng mảng:

```PHP
$query = $request->query();
```

### Nhận dữ liệu với Thuộc tính động

Bạn cũng có thể truy xuất vào input sử dụng thuộc tính động trên đối tượng `Illuminate\Http\Request`. Ví dụ, nếu một trong các form có chứa trường là `name`, bạn có thể lấy giá trị được post lên như thế này:

```PHP
$name = $request->name;
```

Khi sử dụng thuộc tính động, Laravel đầu tiên sẽ tìm giá trị của tham số trên dữ liệu của request và rồi trong route parameter.

### Nhận giá trị input từ JSON

Khi gửi request kiểu JSON lên, bạn có thể truy xuất dữ liệu trong JSON thông qua hàm `input` với điều kiện là header của request `Content-Type` phải được set là `application/json`. Bạn có thể sử dụng cú pháp "dot" để truy xuất sâu hơn vào trong mảng JSON:

```PHP
$name = $request->input('user.name');
```

### Nhận một phần của dữ liệu input

Nếu bạn muốn lấy một tập nhỏ của input, bạn có thể sử dụng hai hàm `only` hay `except`. Cả hai hàm đều nhận một `array` hoặc một danh sách các đối số:

```PHP
$input = $request->only(['username', 'password']);

$input = $request->only('username', 'password');

$input = $request->except(['credit_card']);

$input = $request->except('credit_card');
```

### Kiểm tra một giá trị input có tồn tại

Để kiểm tra một giá trị có tồn tại trên request hay không, bạn có thể sử dụng hàm `has`. Hàm `has` trả về `true` nếu như giá trị tồn tại **và** không phải là chuỗi rỗng:

```PHP
if ($request->has('name')) {
    //
}
```

Khi đưa vào một mảng, phương thức ``has`` sẽ xác định trong tất cả dữ liệu có trong biến:

```PHP
if ($request->has(['name', 'email'])) {
    //
}
```

Nếu bạn muốn kiểm tra xem giá trị có rỗng không, có thể sử dụng phương thức ``filled``:

```PHP
if ($request->filled('name')) {
    //
}
```

### Input cũ

Laravel cho phép bạn giữ giá trị input từ một request sang request tiếp theo. Đặc điểm này đặc biệt hữu dụng khi bạn muốn thiết lập lại form sau khi phát hiện có lỗi. Tuy nhiên, nếu bạn sử dụng [validation services](validation.md) của Laravel, thì bạn không cần phải làm việc này vì các công cụ xử lý validation sẵn của Laravel đã tự động thực hiện rồi.

#### Flash input tới session

Hàm `flash` trong `Illuminate\Http\Request` sẽ flash input hiện tại vào trong [session](session.md) nên nó có thể sử dụng trong request tiếp theo của user tới ứng dụng:

```php
$request->flash();
```

Bạn cũng có thể sử dụng `flashOnly` và `flashExcept` để flash một tập nhỏ của dữ liệu request vào trong session:

```php
$request->flashOnly(['username', 'email']);

$request->flashExcept('password');
```

#### Flash input vào trong session rồi chuyển trang

Vì bạn thường muốn flash input cùng với chuyển trang vào trang trước đó, bạn có thể dễ dàng tạo móc nối input vào trong một redirect sử dụng hàm `withInput`:

```php
return redirect('form')->withInput();

return redirect('form')->withInput$request->except('password'));
```

#### Lấy dữ liệu cũ

Để lấy dữ liệu đã flash từ request trước đó, sử dụng hàm `old` của `Request`. Hàm `old` cung cấp một helper tiện ích cho việc lấy dữ liệu đã flash ra khỏi [session](session.md):

```php
$username = $request->old('username');
```

Laravel cũng cung cấp một helper toàn cục `old`. Nếu như bạn muốn hiển thị giá trị input cũ bên trong [Blade template](blade.md), thì sử dụng helper `old` sẽ tiện hơn. Nếu không có input cũ nào tìm thấy thì giá trị `null` sẽ được trả về:

```php
<input type="text" name="username" value="{{ old('username') }}">
```

### Cookies

#### Lấy cookies từ Request

Tất cả các cookies tạo bởi Laravel đều được mã hoá và kí với một mã xác thực, nghĩa là chúng sẽ bị coi là không hợp lệ nếu chúng bị thay đổi phía client. Để lấy cookie từ request, bạn có thể sử dụng hàm `cookie` từ `Illuminate\Http\Request`:

```php
$value = $request->cookie('name');
```

#### Gắn một cookie mới vào Response

Laravel cung cấp hàm `cookie` helper phục vụ như một factory đơn giản để tạo ra một đối tượng từ class `Symfony\Component\HttpFoundation\Cookie`. Những cookies này có thể được gắn với một đối tượng `Illuminate\Http\Response` sử dụng hàm `withCookie`:

```php
$response = new Illuminate\Http\Response('Hello World');

$response->withCookie('name', 'value', $minutes);

return $response;
```

Để tạo một cookie tồn tại lâu (long-lived), tồn tại trong vòng 5 năm, bạn có thể sử dụng hàm `forever` trong factory bằng cách gọi hàm `cookie` helper không có tham số nào, và rồi gọi móc nối với `forever` để tạo:

```php
$response->withCookie(cookie()->forever('name', 'value'));
```

### Tạo Cookie Instances

Nếu bạn muốn tạo một ``Symfony\Component\HttpFoundation\Cookie`` có thể response sau một khoảng thời gian, bạn có thể sử dụng helper global ``cookie``. Khi đó cookie sẽ không gửi lại cho client trừ khi nó được gán vào response instance:

```PHP
$cookie = cookie('name', 'value', $minutes);

return response('Hello World')->cookie($cookie);
```

## Files

### Lấy file được upload

Bạn có thể lấy files uploaded từ một ``Illuminate\Http\Request`` bằng cách sử dụng phương thức  file hoặc sử dụng thuộc tính động. Phương thức ``file`` sẽ trả về một class  ``Illuminate\Http\UploadedFile``, nó kế thừa từ ``SplFileInfo`` class của PHP và cung cấp một số phương thức để tương tác với fiel:

```PHP
$file = $request->file('photo');

$file = $request->photo;
```

Bạn có thể kiểm tra nếu một file tồn tại trên request sử dụng hàm `hasFile`:

```PHP
if ($request->hasFile('photo')) {
  //
}
```

### Kiểm tra upload thành công

Ngoài việc kiểm tra nếu file upload tồn tại, bạn có thể kiểm tra xem có vấn đề gì trong quá trình file upload lên không thông qua việc sử dụng hàm `isValid`:

```PHP
if ($request->file('photo')->isValid()) {
    //
}
```

### Đường dẫn File & Extensions

Class ``UploadedFile`` ngoài ra còn chứa phương thức lấy đường dẫn đầy đủ và extension của file. Phương thức extension sẽ cho phép đoán extension trên dựa nội dung của file. Extension này có thể khác với extension được cung cấp bởi client:

```PHP
$path = $request->photo->path();

$extension = $request->photo->extension();
```

### Phương thức khác của File

Có một số phương thức tồn tại trong class ``UploadedFile``. Chi tiết xem tại [tài liệu API](http://api.symfony.com/3.0/Symfony/Component/HttpFoundation/File/UploadedFile.html) của class để biết thêm chi tiết các phương thức đấy.

### Chuyển vị trí file upload

Để lưu một file uploaded, thông thường sử dụng một trong những cấu hình filesystems. Class  ``UploadedFile`` có phương thức  ``store`` nó sẽ chuyển file upload từ ổ cứng của bạn đến một nơi có thể là trên local của bạn hoặc ngay cả trên cloud storage như Amazon S3.

Phương thức ``store`` chấp nhận đường dẫn file nên được lưu trữ đường dẫn tương đối so với thư mục gốc cấu hình của filesystem. Đường dẫn không được chứa tên file, tên sẽ tự động được sinh ra bằng cách sử dụng mã hóa MD5 của nội dung file.

Phương thức ``store`` ngoài ra còn chấp nhận tham số thứ hai có tên của nơi mà bạn sử dụng để lưu file. Phương thức sẽ trả về đường dẫn tương đối của file đối với thư mục gốc:

```PHP
$request->file('photo')->move($destinationPath);

$request->file('photo')->move($destinationPath, $fileName);
```

Nếu bạn không muốn tên file được tự động tạo ra, bạn có thể sử dụng phương thứcstoreAs, nó sẽ chấp nhận các đối số như đường dẫn, tên file, và tên nơi lưu:

```PHP
$path = $request->photo->storeAs('images', 'filename.jpg');

$path = $request->photo->storeAs('images', 'filename.jpg', 's3');
```

### Các phương thức khác với file

Có nhiều hàm khác hỗ trợ cho việc xử lý file trong `UploadedFile`. Hãy tham khảo [tài liệu](http://api.symfony.com/3.0/Symfony/Component/HttpFoundation/File/UploadedFile.html) để biết thêm thông tin về các hàm đó.

## Configuring Trusted Proxies

When running your applications behind a load balancer that terminates TLS / SSL certificates, you may notice your application sometimes does not generate HTTPS links. Typically this is because your application is being forwarded traffic from your load balancer on port 80 and does not know it should generate secure links.

To solve this, you may use the ``App\Http\Middleware\TrustProxies`` middleware that is included in your Laravel application, which allows you to quickly customize the load balancers or proxies that should be trusted by your application. Your trusted proxies should be listed as an array on the $proxies property of this middleware. In addition to configuring the trusted proxies, you may configure the proxy $headers that should be trusted:

```PHP
<?php

namespace App\Http\Middleware;

use Illuminate\Http\Request;
use Fideloper\Proxy\TrustProxies as Middleware;

class TrustProxies extends Middleware
{
    /**
     * The trusted proxies for this application.
     *
     * @var array
     */
    protected $proxies = [
        '192.168.1.1',
        '192.168.1.2',
    ];

    /**
     * The headers that should be used to detect proxies.
     *
     * @var string
     */
    protected $headers = Request::HEADER_X_FORWARDED_ALL;
}
```

If you are using AWS Elastic Load Balancing, your $headers value should be  Request::HEADER_X_FORWARDED_AWS_ELB. For more information on the constants that may be used in the $headers property, check out Symfony's documentation on trusting proxies.

### Trusting All Proxies

If you are using Amazon AWS or another "cloud" load balancer provider, you may not know the IP addresses of your actual balancers. In this case, you may use * to trust all proxies:

```PHP
/**
 * The trusted proxies for this application.
 *
 * @var array
 */
protected $proxies = '*';
```


text:
class Gem::Resolver::InstallerSet < Gem::Resolver::Set

  ##
  # List of Gem::Specification objects that must always be installed.

  attr_reader :always_install # :nodoc:

  ##
  # Only install gems in the always_install list

  attr_accessor :ignore_dependencies # :nodoc:

  ##
  # Do not look in the installed set when finding specifications.  This is
  # used by the --install-dir option to `gem install`

  attr_accessor :ignore_installed # :nodoc:

  ##
  # Creates a new InstallerSet that will look for gems in +domain+.

  def initialize domain
    @domain = domain

    @f = Gem::SpecFetcher.fetcher

    @all = Hash.new { |h,k| h[k] = [] }
    @always_install      = []
    @ignore_dependencies = false
    @ignore_installed    = false
    @loaded_remote_specs = []
    @specs               = {}
  end

  ##
  # Should local gems should be considered?

  def consider_local? # :nodoc:
    @domain == :both or @domain == :local
  end

  ##
  # Should remote gems should be considered?

  def consider_remote? # :nodoc:
    @domain == :both or @domain == :remote
  end

  ##
  # Returns an array of IndexSpecification objects matching DependencyRequest
  # +req+.

  def find_all req
    res = []

    dep  = req.dependency

    return res if @ignore_dependencies and
              @always_install.none? { |spec| dep.matches_spec? spec }

    name = dep.name

    dep.matching_specs.each do |gemspec|
      next if @always_install.include? gemspec

      res << Gem::Resolver::InstalledSpecification.new(self, gemspec)
    end unless @ignore_installed

    if consider_local? then
      local_source = Gem::Source::Local.new

      if spec = local_source.find_gem(name, dep.requirement) then
        res << Gem::Resolver::IndexSpecification.new(
          self, spec.name, spec.version, local_source, spec.platform)
      end
    end

    if consider_remote? then
      load_remote_specs dep

      @all[name].each do |remote_source, n|
        if dep.match? n then
          res << Gem::Resolver::IndexSpecification.new(
            self, n.name, n.version, remote_source, n.platform)
        end
      end
    end

    res
  end

  def inspect # :nodoc:
    always_install = @always_install.map { |s| s.full_name }

    '#<%s domain: %s specs: %p always install: %p>' % [
      self.class, @domain, @specs.keys, always_install,
    ]
  end

  ##
  # Loads remote prerelease specs if +dep+ is a prerelease dependency

  def load_remote_specs dep # :nodoc:
    types = [:released]
    types << :prerelease if dep.prerelease?

    types.each do |type|
      next if @loaded_remote_specs.include? type
      @loaded_remote_specs << type

      list, = @f.available_specs type

      list.each do |uri, specs|
        specs.each do |n|
          @all[n.name] << [uri, n]
        end
      end
    end
  end

  ##
  # Called from IndexSpecification to get a true Specification
  # object.

  def load_spec name, ver, platform, source # :nodoc:
    key = "#{name}-#{ver}-#{platform}"

    @specs.fetch key do
      tuple = Gem::NameTuple.new name, ver, platform

      @specs[key] = source.fetch_spec tuple
    end
  end

  def pretty_print q # :nodoc:
    q.group 2, '[InstallerSet', ']' do
      q.breakable
      q.text "domain: #{@domain}"

      q.breakable
      q.text 'specs: '
      q.pp @specs.keys

      q.breakable
      q.text 'always install: '
      q.pp @always_install
    end
  end

end


text:
@interface CandidateTableViewController()
@end

@implementation CandidateTableViewController{
    __weak UIPopoverController *screeningPopover;
    __weak UIPopoverController *majorPopover;
    __weak UIPopoverController *locationPopover;
    __weak UIPopoverController *gradDatePopover;
    __weak UIPopoverController *techSkillsPopover;
    __weak UIPopoverController *teamPopover;
}

#pragma mark - Properties
@synthesize nameTextField;
@synthesize overallGPATextField;
@synthesize majorGPATextField;
@synthesize emailAddressField;
@synthesize notesTextView;
@synthesize homeBarButton;

@synthesize tempScreeningDocument = _tempScreeningDocument;
@synthesize majorLabel = _majorLabel;
@synthesize locationLabel;
@synthesize gradDateLabel;
@synthesize sponsorshipControl;
@synthesize levelOfEducationControl;
@synthesize candidate = _candidate;
@synthesize screenDoc = _screening;

#pragma mark - Instance Variables
PlaceholderView *_placeholder;
NSString *_currentScreeningFilename;
MBProgressHUD *progress;

- (IBAction)homeBarButtonPressed:(UIBarButtonItem *)sender {
    [TestFlight passCheckpoint:@"Returning Home from Candidate"];
    progress = [MBProgressHUD showHUDAddedTo:self.view animated:YES];
    progress.mode = MBProgressHUDModeIndeterminate;
    progress.labelText = @"Saving Screening";
    progress.removeFromSuperViewOnHide = YES;

    [self.screenDoc saveToURL:self.screenDoc.fileURL forSaveOperation:UIDocumentSaveForOverwriting completionHandler:^(BOOL success) {
        progress.hidden = YES;
        if(success)
        {
            NSLog(@"Document was successfully saved before returning home");
            [self performSegueWithIdentifier:@"ReturnHome" sender:self];
        }else{
            NSLog(@"Document failed to save on returning home");
            UIAlertView *errorAlert = [[UIAlertView alloc] initWithTitle:@"Proglem" message:@"There was a problem saving your screening.  If this continues to happen, please contact support." delegate:self cancelButtonTitle:nil otherButtonTitles:nil];
            [errorAlert show];
        }
    }];

}

- (void)newScreeningSelected
{
    NSLog(@"New Screening selected");
}

- (IBAction)showScreeningPopover:(id)sender{
    if (screeningPopover)
        [screeningPopover dismissPopoverAnimated:YES];
    else
        [self performSegueWithIdentifier:@"ScreeningTableViewControllerPopover" sender:sender];
}

- (void)prepareMajorPopover:(UIStoryboardSegue *)segue {
    //Fetch all majors in document;
    NSFetchRequest *majorRequest = [NSFetchRequest fetchRequestWithEntityName:@"Major"];
    NSSortDescriptor *sortDescriptor = [[NSSortDescriptor alloc] initWithKey:@"name" ascending:YES];

    majorRequest.sortDescriptors = [NSArray arrayWithObject:sortDescriptor];

    NSError *error;
    NSArray *majorArray = [self.screenDoc.managedObjectContext executeFetchRequest:majorRequest error:&error];

    //get destination of major view controller
    UINavigationController *majorNav = (UINavigationController *)[segue destinationViewController];
    MajorViewController *majorViewController = [majorNav.viewControllers objectAtIndex:0];
    //register for delegate
    majorViewController.delegate = self;

    //set done button to disabled if no values in array;
    if(majorArray.count < 1){
        majorViewController.navigationItem.rightBarButtonItem.enabled = NO;
    }else {
        majorViewController.navigationItem.rightBarButtonItem.enabled = YES;

        NSPredicate *filter = [NSPredicate predicateWithFormat:@"self.name == %@",self.majorLabel.text];
        NSArray *filteredScreenings = [majorArray filteredArrayUsingPredicate:filter];

        majorViewController.selectedValue = [filteredScreenings lastObject];
    }
    //set major array in popover class
    majorViewController.majorArray = majorArray;

    //weak reference to segue to prevent from repopping
    majorPopover = [(UIStoryboardPopoverSegue *)segue popoverController];
}

- (void)prepareLocationPopover:(UIStoryboardSegue *)segue {
    //Fetch all locations in document
    NSFetchRequest *locationRequest = [NSFetchRequest fetchRequestWithEntityName:@"Location"];

    NSSortDescriptor *sortDescriptor = [[NSSortDescriptor alloc] initWithKey:@"name" ascending:YES];
    locationRequest.sortDescriptors = [NSArray arrayWithObject:sortDescriptor];

    NSError *error;
    NSArray *locationArray = [self.screenDoc.managedObjectContext executeFetchRequest:locationRequest error:&error];
    NSLog(@"Size of locationArray = %d", locationArray.count);

    UINavigationController *locationNav = (UINavigationController *)[segue destinationViewController];
    LocationPickerViewController *locationViewController = [locationNav.viewControllers objectAtIndex:0];
    locationViewController.delegate = self;

    //set done button to disabled if no values in array;
    if(locationArray.count < 1){
        locationViewController.navigationItem.rightBarButtonItem.enabled = NO;
    }else {
        locationViewController.navigationItem.rightBarButtonItem.enabled = YES;

        NSPredicate *filter = [NSPredicate predicateWithFormat:@"self.name == %@",self.locationLabel.text];
        NSArray *filteredScreenings = [locationArray filteredArrayUsingPredicate:filter];

        locationViewController.selectedValue = [filteredScreenings lastObject];
    }

    locationViewController.locationArray = locationArray;

    locationPopover = [(UIStoryboardPopoverSegue *) segue popoverController];
}

- (void)prepareGradDatePopover:(UIStoryboardSegue *)segue {
    UINavigationController *gradDateNav = (UINavigationController *)[segue destinationViewController];
    GradDateViewController *gradDateViewController = [gradDateNav.viewControllers objectAtIndex:0];
    gradDateViewController.delegate = self;

    if (self.candidate.graduationDate != nil) {
        NSLog(@"the current grad date is %@",self.candidate.graduationDate);
        [gradDateViewController.gradDatePicker setDate:self.candidate.graduationDate];
    }

    gradDatePopover = [(UIStoryboardPopoverSegue *) segue popoverController];
}

- (void)prepareTechnicalSkillsPopover:(UIStoryboardSegue *)segue {
    //Fetch all locations in document
    NSFetchRequest *techSkillsRequest = [NSFetchRequest fetchRequestWithEntityName:@"TechnicalSkill"];
    NSError *error;
    NSArray *techSkillsArray = [self.screenDoc.managedObjectContext executeFetchRequest:techSkillsRequest error:&error];
    NSLog(@"Size of techskillsArray = %d", techSkillsArray.count);

    UINavigationController *techSkillsNav = (UINavigationController *)[segue destinationViewController];
    TechnicalSkillsTableViewController *techSkillsViewController = [techSkillsNav.viewControllers objectAtIndex:0];
    techSkillsViewController.delegate = self;
    techSkillsViewController.technicalSkillsArray = techSkillsArray;

    techSkillsPopover = [(UIStoryboardPopoverSegue *) segue popoverController];
}

- (void)prepareReturnHome {

}

- (void)preparePushQuestion:(UIStoryboardSegue *)segue {
    QuestionsTableViewController *questionVc = (QuestionsTableViewController *)[segue destinationViewController];
    questionVc.screening = self.screenDoc;
}

// FIXME: Refractor this method
- (void)prepareForSegue:(UIStoryboardSegue *)segue sender:(id)sender{

    [self resignAllFirstResponders];

    if([[segue identifier] isEqualToString:@"PushQuestion"]){
        [self preparePushQuestion:segue];
    }

    if([[segue identifier] isEqualToString:@"MajorPopover"]){
        [self prepareMajorPopover:segue];
    }

    if([[segue identifier] isEqualToString:@"LocationPopover"]){
        [self prepareLocationPopover:segue];
    }

    if([[segue identifier] isEqualToString:@"GradDatePopover"]){

        [self prepareGradDatePopover:segue];
    }

    if([[segue identifier] isEqualToString:@"TechnicalSkillsPopover"]){

        [self prepareTechnicalSkillsPopover:segue];
    }

    if([[segue identifier] isEqualToString:@"ReturnHome"]){
        [self prepareReturnHome];
    }

    //do pre stuff for
}

#pragma mark - Instance Methods

- (void)showPlaceholderAnimated:(BOOL)animated
{
    if(!_placeholder)
        _placeholder = [[PlaceholderView alloc] initWithFrame:[[UIScreen mainScreen] bounds] MessageText:@"Please select or create a new screening."];
    if(animated){
        _placeholder.alpha = 0.0;
        [self.view addSubview:_placeholder];
        [UIView animateWithDuration:0.3 animations:^{
            _placeholder.alpha = 1.0;
        }];
    }else{
        [self.view addSubview:_placeholder];
    }
}

- (void)checkValidationStatus
{
    if(self.candidate.major && self.candidate.name && self.candidate.emailAddress.length > 0){
        self.navigationItem.rightBarButtonItem.enabled = YES;
    }
}

- (void)loadCurrentValues
{
    [TestFlight passCheckpoint:@"Loading current values for screen document"];
    // have property on candidate view controller
    NSError *error;
    NSFetchRequest *fetchRequest = [NSFetchRequest fetchRequestWithEntityName:@"Candidate"];

    NSArray *candidateArray = [self.screenDoc.managedObjectContext executeFetchRequest:fetchRequest error:&error];

    self.candidate = candidateArray.lastObject;

    // Set up fields with values from candidate
    self.nameTextField.text = self.candidate.name;

    NSLog(@"Candidate name is set to : %@", self.candidate.name);

    NSNumberFormatter *number = [[NSNumberFormatter alloc] init];
    number.numberStyle = NSNumberFormatterDecimalStyle;

    if([self.candidate.overallGpa doubleValue] > 0.0){
        self.overallGPATextField.text = [number stringFromNumber:self.candidate.overallGpa];
        self.overallGPATextField.textColor = [UIColor blackColor];
    }

    if([self.candidate.majorGpa doubleValue] > 0.0){
        self.majorGPATextField.text = [number stringFromNumber:self.candidate.majorGpa];
        self.majorGPATextField.textColor = [UIColor blackColor];
    }

    if(self.candidate.major.name){
        self.majorLabel.text = self.candidate.major.name;
        self.majorLabel.textColor = [UIColor blackColor];
    }

    if(self.candidate.notes){
        self.notesTextView.text = self.candidate.notes;
        self.notesTextView.textColor = [UIColor blackColor];
    }

    if(self.candidate.location.name){
        self.locationLabel.text = self.candidate.location.name;
        self.locationLabel.textColor = [UIColor blackColor];
    }

    if(self.candidate.graduationDate){

        NSDateFormatter *formatter = [[NSDateFormatter alloc] init];
        [formatter setDateFormat:@"MM/yyyy"];

        NSString *stringFromDate = [formatter stringFromDate:self.candidate.graduationDate];
        self.gradDateLabel.text = stringFromDate;
        self.gradDateLabel.textColor = [UIColor blackColor];
    }

    if(self.candidate.emailAddress){
        self.emailAddressField.text = self.candidate.emailAddress;
        self.emailAddressField.textColor = [UIColor blackColor];
    }

    if(self.candidate.major.levelOfEducation){
        if([self.candidate.major.levelOfEducation isEqualToString:@"Doctorate"]){
            self.levelOfEducationControl.selectedSegmentIndex = 2;
        }else if([self.candidate.major.levelOfEducation isEqualToString:@"Master"]){
            self.levelOfEducationControl.selectedSegmentIndex = 1;
        }else{
        self.levelOfEducationControl.selectedSegmentIndex = 0;
        }

    }

    self.sponsorshipControl.selectedSegmentIndex = [self.candidate.sponsorshipNeeded intValue];

    [self checkValidationStatus];
}

- (void)performPushQuestionSegue
{
    [self performSegueWithIdentifier:@"PushQuestion" sender:self];
}

-(void)dismissKeyboard
{
    [self.view endEditing:YES];
}

#pragma mark - View lifecycle

- (void)viewDidLoad
{
    [super viewDidLoad];

    // Adjust UITextField alignment for 6.0
    //self.nameTextField.textAlignment = UITextAlignmentLeft;
    //self.nameTextField.textAlignment = UITextAlignmentRight;

    //self.emailAddressField.textAlignment = UITextAlignmentLeft;
    //self.emailAddressField.textAlignment = UITextAlignmentRight;

    //self.overallGPATextField.textAlignment = UITextAlignmentLeft;
    //self.overallGPATextField.textAlignment = UITextAlignmentRight;

    //self.majorGPATextField.textAlignment = UITextAlignmentLeft;
    //self.majorGPATextField.textAlignment = UITextAlignmentRight;

    self.tableView.backgroundView = [[UIView alloc] initWithFrame:self.tableView.bounds];
    //self.tableView.backgroundView.backgroundColor = [UIColor colorWithPatternImage:[UIImage imageNamed:@"book-cover-011.png"]];

    [[NSNotificationCenter defaultCenter] addObserver:self
                                             selector:@selector(documentStateChanged:)
                                                 name:UIDocumentStateChangedNotification
                                               object:self.screenDoc];

    // Create custom 'forward' button and add it to the nav control
    UIButton *forwardButton = [UIButton buttonWithType:101];
    forwardButton.transform = CGAffineTransformMakeScale(-1,1);
    forwardButton.titleLabel.transform = CGAffineTransformMakeScale(-1,1);
    [forwardButton setTitle:@"Questions" forState:UIControlStateNormal];
    [forwardButton addTarget:self action:@selector(performPushQuestionSegue) forControlEvents:UIControlEventTouchUpInside];

    UIBarButtonItem *questionsButton = [[UIBarButtonItem alloc] initWithCustomView:forwardButton];
    self.navigationItem.rightBarButtonItem = questionsButton;

    questionsButton.enabled = NO;

    // Add GestureRecognizer to allow for tapping the background to dismiss keyboard
    UITapGestureRecognizer *tap = [[UITapGestureRecognizer alloc] initWithTarget:self
                                                                          action:@selector(dismissKeyboard)];
    tap.cancelsTouchesInView = NO;
    [self.view addGestureRecognizer:tap];

    [self loadCurrentValues];

//    if(!self.screening){
//        [self showPlaceholderAnimated:NO];
//    }

}

- (void)viewDidUnload
{
    [self setNameTextField:nil];
    [self setOverallGPATextField:nil];
    [self setMajorGPATextField:nil];
    [self setLocationLabel:nil];
    [self setGradDateLabel:nil];
    [self setSponsorshipControl:nil];

    [[NSNotificationCenter defaultCenter] removeObserver:self];

    [self setEmailAddressField:nil];
    [self setNotesTextView:nil];
    [self setHomeBarButton:nil];
    [self setNotesTextView:nil];
    [self setLevelOfEducationControl:nil];
    [super viewDidUnload];
    // Release any retained subviews of the main view.
    // e.g. self.myOutlet = nil;

}

- (void)viewWillAppear:(BOOL)animated
{
    [super viewWillAppear:animated];
}

- (void)viewDidAppear:(BOOL)animated
{
    [super viewDidAppear:animated];

}

- (void)viewWillDisappear:(BOOL)animated
{
	[super viewWillDisappear:animated];
}

- (void)viewDidDisappear:(BOOL)animated
{
	[super viewDidDisappear:animated];
}

- (BOOL)shouldAutorotateToInterfaceOrientation:(UIInterfaceOrientation)interfaceOrientation
{
    // Only support Portrait
    if (interfaceOrientation == UIInterfaceOrientationPortrait)
        return YES;
    else
        return NO;
}

- (void)didReceiveMemoryWarning
{
    [super didReceiveMemoryWarning];
    // Release any cached data, images, etc that aren't in use.
}

#pragma mark - Table View Delegate

- (UIView *)tableView:(UITableView *)tableView viewForHeaderInSection:(NSInteger)section {
    ShadowTableHeaderView *header = [[ShadowTableHeaderView alloc] init];
    header.titleLabel.text = [self tableView:tableView titleForHeaderInSection:section];

    return header;
}

- (CGFloat)tableView:(UITableView *)tableView heightForHeaderInSection:(NSInteger)section
{
    return 50.0;
}

- (UIView *)tableView:(UITableView *)tableView viewForFooterInSection:(NSInteger)section
{
    ShadowTableFooterView *footer = [[ShadowTableFooterView alloc] init];
    return footer;
}

- (CGFloat)tableView:(UITableView *)tableView heightForFooterInSection:(NSInteger)section
{
    return 20.0;
}

#pragma mark UITextFieldDelegate Protocol

- (BOOL)textFieldShouldReturn:(UITextField *)textField {
    [textField resignFirstResponder];
    return NO;
}

#pragma mark LoginModalProtocal

- (void)loginDidFinishWithUsername:(NSString *)username EventName:(NSString *)eventName
{
    // Set session
    AppDelegate *appDelegate = (AppDelegate *)[[UIApplication sharedApplication] delegate];

    appDelegate.session = username;

    // Dismiss Modal
    [self dismissModalViewControllerAnimated:YES];

    NSLog(@"Login Finished.  Username: %@ Event Name: %@",username,eventName);
}

#pragma mark Major Protocal

- (void)majorWasSelected:(Major *)major
{
    self.majorLabel.text = major.name;
    self.majorLabel.textColor = [UIColor blackColor];
    if(majorPopover)
        [majorPopover dismissPopoverAnimated:YES];
    self.candidate.major = major;
    [self.candidate removeQuestions:self.candidate.questions];
    [self.candidate populateTechnicalQuestions];
    [self checkValidationStatus];
}

#pragma mark Location Protocal
-(void)locationWasSelected:(Location *)location
{
    self.locationLabel.text = location.name;
    self.locationLabel.textColor = [UIColor blackColor];
    if(locationPopover)
        [locationPopover dismissPopoverAnimated:YES];
    self.candidate.location = location;
}

#pragma mark GradDate Protocal
-(void)gradDateSelected:(NSDate *)gradDate
{
    NSDateFormatter *formatter = [[NSDateFormatter alloc] init];
    [formatter setDateFormat:@"MM/yyyy"];

    NSString *stringFromDate = [formatter stringFromDate:gradDate];
    self.gradDateLabel.text = stringFromDate;
    self.gradDateLabel.textColor =[UIColor blackColor];

    if(gradDatePopover)
        [gradDatePopover dismissPopoverAnimated:YES];
    self.candidate.graduationDate = gradDate;

}

#pragma mark TechSkills Protocal
-(void)TechnicalSkillsWereSelected:(NSArray *)technicalSkills
{
    if(techSkillsPopover)
        [techSkillsPopover dismissPopoverAnimated:YES];
#warning TODO: set value to candidate
    for (id key in technicalSkills) {
        NSLog(@"key: %@, value: %@ \n", key, [technicalSkills objectAtIndex:key]);
    }
}

#pragma mark Team Protocal
-(void)teamsWereSelected:(NSArray *)teams
{
    if(teamPopover)
        [teamPopover dismissPopoverAnimated:YES];
}

#pragma mark Candidate Name and GPAs
- (void)textFieldDidEndEditing:(UITextField *)textField{
    //check twhich text field sent message
    if ([textField isEqual:self.nameTextField]) {
        //set candidatename
        self.candidate.name = self.nameTextField.text;
        [self checkValidationStatus];
    }
    else if([textField isEqual:self.emailAddressField]){
     //set email
        self.candidate.emailAddress = self.emailAddressField.text;
        [self checkValidationStatus];
    }else if ([textField isEqual:self.majorGPATextField]) {
        //CONVERT NSString to NSNumber
        NSNumberFormatter *f = [[NSNumberFormatter alloc] init];
        [f setNumberStyle:NSNumberFormatterDecimalStyle];
        NSNumber * number = [f numberFromString:self.majorGPATextField.text];
        //setmajorgpa on candidate
        self.candidate.majorGpa = number;
    }else if ([textField isEqual:self.overallGPATextField]) {
        //CONVERT NSString to NSNumber
        NSNumberFormatter *f = [[NSNumberFormatter alloc] init];
        [f setNumberStyle:NSNumberFormatterDecimalStyle];
        NSNumber * number = [f numberFromString:self.overallGPATextField.text];
        //set overall gpa on candidate
        self.candidate.overallGpa = number;
    }
}

- (void) textViewDidEndEditing:(UITextView *)textView{
    if ([textView isEqual:self.notesTextView]) {
        self.candidate.notes = self.notesTextView.text;
    }
}

#pragma mark SponorshipNeeded
- (IBAction)sponsorshipValueChanged:(UISegmentedControl *)sponsorControl
{
    if ([[sponsorControl titleForSegmentAtIndex:sponsorControl.selectedSegmentIndex] isEqualToString:@"Yes"]) {
        self.candidate.sponsorshipNeeded = [NSNumber numberWithBool:YES];
    }
    else {
        self.candidate.sponsorshipNeeded = [NSNumber numberWithBool:NO];
;
    }
}

#pragma mark LevelOfEducation
- (IBAction)levelOfEducationValueChanged:(UISegmentedControl *)educationControl
{
    self.candidate.major.levelOfEducation = [educationControl titleForSegmentAtIndex:educationControl.selectedSegmentIndex];
}

-(void)resignAllFirstResponders{
    [self.nameTextField resignFirstResponder];
    [self.overallGPATextField resignFirstResponder];
    [self.majorLabel resignFirstResponder];
    [self.emailAddressField resignFirstResponder];
}

#pragma mark - Notifications

- (void)documentStateChanged:(NSNotification *)notification
{
    ScreenDocument *screenDocument = (ScreenDocument *)[notification object];
    if(screenDocument.documentState == UIDocumentStateClosed){
        NSLog(@"Document is closed, should open it");
    }else if(screenDocument.documentState == UIDocumentStateSavingError){
        NSLog(@"There was a problem saving the document");
    }else if(screenDocument.documentState == UIDocumentStateInConflict){
        NSLog(@"Document is reporting a conflict");
    }else if(screenDocument.documentState == UIDocumentStateNormal){
        NSLog(@"Document is ready!");
        //[self screenDocumentIsReady];
    }
}

@end