text:
package org.apache.flink.core.fs;

import org.apache.flink.annotation.Internal;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.configuration.IllegalConfigurationException;
import org.apache.flink.util.function.SupplierWithException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;

import java.io.Closeable;
import java.io.IOException;
import java.net.URI;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;

/**
 * A file system that limits the number of concurrently open input streams,
 * output streams, and total streams for a target file system.
 *
 *
This file system can wrap another existing file system in cases where
 * the target file system cannot handle certain connection spikes and connections
 * would fail in that case. This happens, for example, for very small HDFS clusters
 * with few RPC handlers, when a large Flink job tries to build up many connections during
 * a checkpoint.
 *
 *
The filesystem may track the progress of streams and close streams that have been
 * inactive for too long, to avoid locked streams of taking up the complete pool.
 * Rather than having a dedicated reaper thread, the calls that try to open a new stream
 * periodically check the currently open streams once the limit of open streams is reached.
 */
@Internal
public class LimitedConnectionsFileSystem extends FileSystem {

	private static final Logger LOG = LoggerFactory.getLogger(LimitedConnectionsFileSystem.class);

	/** The original file system to which connections are limited. */
	private final FileSystem originalFs;

	/** The lock that synchronizes connection bookkeeping. */
	private final ReentrantLock lock;

	/** Condition for threads that are blocking on the availability of new connections. */
	private final Condition available;

	/** The maximum number of concurrently open output streams. */
	private final int maxNumOpenOutputStreams;

	/** The maximum number of concurrently open input streams. */
	private final int maxNumOpenInputStreams;

	/** The maximum number of concurrently open streams (input + output). */
	private final int maxNumOpenStreamsTotal;

	/** The nanoseconds that a opening a stream may wait for availability. */
	private final long streamOpenTimeoutNanos;

	/** The nanoseconds that a stream may spend not writing any bytes before it is closed as inactive. */
	private final long streamInactivityTimeoutNanos;

	/** The set of currently open output streams. */
	@GuardedBy("lock")
	private final HashSet<OutStream> openOutputStreams;

	/** The set of currently open input streams. */
	@GuardedBy("lock")
	private final HashSet<InStream> openInputStreams;

	/** The number of output streams reserved to be opened. */
	@GuardedBy("lock")
	private int numReservedOutputStreams;

	/** The number of input streams reserved to be opened. */
	@GuardedBy("lock")
	private int numReservedInputStreams;

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

	/**
	 * Creates a new output connection limiting file system.
	 *
	 *
If streams are inactive (meaning not writing bytes) for longer than the given timeout,
	 * then they are terminated as "inactive", to prevent that the limited number of connections gets
	 * stuck on only blocked threads.
	 *
	 * @param originalFs              The original file system to which connections are limited.
	 * @param maxNumOpenStreamsTotal  The maximum number of concurrent open streams (0 means no limit).
	 */
	public LimitedConnectionsFileSystem(FileSystem originalFs, int maxNumOpenStreamsTotal) {
		this(originalFs, maxNumOpenStreamsTotal, 0, 0);
	}

	/**
	 * Creates a new output connection limiting file system.
	 *
	 *
If streams are inactive (meaning not writing bytes) for longer than the given timeout,
	 * then they are terminated as "inactive", to prevent that the limited number of connections gets
	 * stuck on only blocked threads.
	 *
	 * @param originalFs              The original file system to which connections are limited.
	 * @param maxNumOpenStreamsTotal  The maximum number of concurrent open streams (0 means no limit).
	 * @param streamOpenTimeout       The maximum number of milliseconds that the file system will wait when
	 *                                no more connections are currently permitted.
	 * @param streamInactivityTimeout The milliseconds that a stream may spend not writing any
	 *                                bytes before it is closed as inactive.
	 */
	public LimitedConnectionsFileSystem(
			FileSystem originalFs,
			int maxNumOpenStreamsTotal,
			long streamOpenTimeout,
			long streamInactivityTimeout) {
		this(originalFs, maxNumOpenStreamsTotal, 0, 0, streamOpenTimeout, streamInactivityTimeout);
	}

	/**
	 * Creates a new output connection limiting file system, limiting input and output streams with
	 * potentially different quotas.
	 *
	 *
If streams are inactive (meaning not writing bytes) for longer than the given timeout,
	 * then they are terminated as "inactive", to prevent that the limited number of connections gets
	 * stuck on only blocked threads.
	 *
	 * @param originalFs              The original file system to which connections are limited.
	 * @param maxNumOpenStreamsTotal  The maximum number of concurrent open streams (0 means no limit).
	 * @param maxNumOpenOutputStreams The maximum number of concurrent open output streams (0 means no limit).
	 * @param maxNumOpenInputStreams  The maximum number of concurrent open input streams (0 means no limit).
	 * @param streamOpenTimeout       The maximum number of milliseconds that the file system will wait when
	 *                                no more connections are currently permitted.
	 * @param streamInactivityTimeout The milliseconds that a stream may spend not writing any
	 *                                bytes before it is closed as inactive.
	 */
	public LimitedConnectionsFileSystem(
			FileSystem originalFs,
			int maxNumOpenStreamsTotal,
			int maxNumOpenOutputStreams,
			int maxNumOpenInputStreams,
			long streamOpenTimeout,
			long streamInactivityTimeout) {

		checkArgument(maxNumOpenStreamsTotal >= 0, "maxNumOpenStreamsTotal must be >= 0");
		checkArgument(maxNumOpenOutputStreams >= 0, "maxNumOpenOutputStreams must be >= 0");
		checkArgument(maxNumOpenInputStreams >= 0, "maxNumOpenInputStreams must be >= 0");
		checkArgument(streamOpenTimeout >= 0, "stream opening timeout must be >= 0 (0 means infinite timeout)");
		checkArgument(streamInactivityTimeout >= 0, "stream inactivity timeout must be >= 0 (0 means infinite timeout)");

		this.originalFs = checkNotNull(originalFs, "originalFs");
		this.lock = new ReentrantLock(true);
		this.available = lock.newCondition();
		this.openOutputStreams = new HashSet<>();
		this.openInputStreams = new HashSet<>();
		this.maxNumOpenStreamsTotal = maxNumOpenStreamsTotal;
		this.maxNumOpenOutputStreams = maxNumOpenOutputStreams;
		this.maxNumOpenInputStreams = maxNumOpenInputStreams;

		// assign nanos overflow aware
		final long openTimeoutNanos = streamOpenTimeout * 1_000_000;
		final long inactivityTimeoutNanos = streamInactivityTimeout * 1_000_000;

		this.streamOpenTimeoutNanos =
				openTimeoutNanos >= streamOpenTimeout ? openTimeoutNanos : Long.MAX_VALUE;

		this.streamInactivityTimeoutNanos =
				inactivityTimeoutNanos >= streamInactivityTimeout ? inactivityTimeoutNanos : Long.MAX_VALUE;
	}

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

	/**
	 * Gets the maximum number of concurrently open output streams.
	 */
	public int getMaxNumOpenOutputStreams() {
		return maxNumOpenOutputStreams;
	}

	/**
	 * Gets the maximum number of concurrently open input streams.
	 */
	public int getMaxNumOpenInputStreams() {
		return maxNumOpenInputStreams;
	}

	/**
	 * Gets the maximum number of concurrently open streams (input + output).
	 */
	public int getMaxNumOpenStreamsTotal() {
		return maxNumOpenStreamsTotal;
	}

	/**
	 * Gets the number of milliseconds that a opening a stream may wait for availability in the
	 * connection pool.
	 */
	public long getStreamOpenTimeout() {
		return streamOpenTimeoutNanos / 1_000_000;
	}

	/**
	 * Gets the milliseconds that a stream may spend not writing any bytes before it is closed as inactive.
	 */
	public long getStreamInactivityTimeout() {
		return streamInactivityTimeoutNanos / 1_000_000;
	}

	/**
	 * Gets the total number of open streams (input plus output).
	 */
	public int getTotalNumberOfOpenStreams() {
		lock.lock();
		try {
			return numReservedOutputStreams + numReservedInputStreams;
		} finally {
			lock.unlock();
		}
	}

	/**
	 * Gets the number of currently open output streams.
	 */
	public int getNumberOfOpenOutputStreams() {
		lock.lock();
		try {
			return numReservedOutputStreams;
		}
		finally {
			lock.unlock();
		}
	}

	/**
	 * Gets the number of currently open input streams.
	 */
	public int getNumberOfOpenInputStreams() {
		return numReservedInputStreams;
	}

	// ------------------------------------------------------------------------
	//  input & output stream opening methods
	// ------------------------------------------------------------------------

	@Override
	public FSDataOutputStream create(Path f, WriteMode overwriteMode) throws IOException {
		return createOutputStream(() -> originalFs.create(f, overwriteMode));
	}

	@Override
	@Deprecated
	@SuppressWarnings("deprecation")
	public FSDataOutputStream create(
			Path f,
			boolean overwrite,
			int bufferSize,
			short replication,
			long blockSize) throws IOException {

		return createOutputStream(() -> originalFs.create(f, overwrite, bufferSize, replication, blockSize));
	}

	@Override
	public FSDataInputStream open(Path f, int bufferSize) throws IOException {
		return createInputStream(() -> originalFs.open(f, bufferSize));
	}

	@Override
	public FSDataInputStream open(Path f) throws IOException {
		return createInputStream(() -> originalFs.open(f));
	}

	private FSDataOutputStream createOutputStream(
			final SupplierWithException<FSDataOutputStream, IOException> streamOpener) throws IOException {

		final SupplierWithException<OutStream, IOException> wrappedStreamOpener =
				() -> new OutStream(streamOpener.get(), this);

		return createStream(wrappedStreamOpener, openOutputStreams, true);
	}

	private FSDataInputStream createInputStream(
			final SupplierWithException<FSDataInputStream, IOException> streamOpener) throws IOException {

		final SupplierWithException<InStream, IOException> wrappedStreamOpener =
				() -> new InStream(streamOpener.get(), this);

		return createStream(wrappedStreamOpener, openInputStreams, false);
	}

	// ------------------------------------------------------------------------
	//  other delegating file system methods
	// ------------------------------------------------------------------------

	@Override
	public FileSystemKind getKind() {
		return originalFs.getKind();
	}

	@Override
	public boolean isDistributedFS() {
		return originalFs.isDistributedFS();
	}

	@Override
	public Path getWorkingDirectory() {
		return originalFs.getWorkingDirectory();
	}

	@Override
	public Path getHomeDirectory() {
		return originalFs.getHomeDirectory();
	}

	@Override
	public URI getUri() {
		return originalFs.getUri();
	}

	@Override
	public FileStatus getFileStatus(Path f) throws IOException {
		return originalFs.getFileStatus(f);
	}

	@Override
	public BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len) throws IOException {
		return originalFs.getFileBlockLocations(file, start, len);
	}

	@Override
	public FileStatus[] listStatus(Path f) throws IOException {
		return originalFs.listStatus(f);
	}

	@Override
	public boolean delete(Path f, boolean recursive) throws IOException {
		return originalFs.delete(f, recursive);
	}

	@Override
	public boolean mkdirs(Path f) throws IOException {
		return originalFs.mkdirs(f);
	}

	@Override
	public boolean rename(Path src, Path dst) throws IOException {
		return originalFs.rename(src, dst);
	}

	@Override
	public boolean exists(Path f) throws IOException {
		return originalFs.exists(f);
	}

	@Override
	@Deprecated
	@SuppressWarnings("deprecation")
	public long getDefaultBlockSize() {
		return originalFs.getDefaultBlockSize();
	}

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

	private <T extends StreamWithTimeout> T createStream(
			final SupplierWithException<T, IOException> streamOpener,
			final HashSet<T> openStreams,
			final boolean output) throws IOException {

		final int outputLimit = output && maxNumOpenInputStreams > 0 ? maxNumOpenOutputStreams : Integer.MAX_VALUE;
		final int inputLimit = !output && maxNumOpenInputStreams > 0 ? maxNumOpenInputStreams : Integer.MAX_VALUE;
		final int totalLimit = maxNumOpenStreamsTotal > 0 ? maxNumOpenStreamsTotal : Integer.MAX_VALUE;
		final int outputCredit = output ? 1 : 0;
		final int inputCredit = output ? 0 : 1;

		// because waiting for availability may take long, we need to be interruptible here
		// and handle interrupted exceptions as I/O errors
		// even though the code is written to make sure the lock is held for a short time only,
		// making the lock acquisition interruptible helps to guard against the cases where
		// a supposedly fast operation (like 'getPos()' on a stream) actually takes long.
		try {
			lock.lockInterruptibly();
			try {
				// some integrity checks
				assert openOutputStreams.size() <= numReservedOutputStreams;
				assert openInputStreams.size() <= numReservedInputStreams;

				// wait until there are few enough streams so we can open another
				waitForAvailability(totalLimit, outputLimit, inputLimit);

				// We do not open the stream here in the locked scope because opening a stream
				// could take a while. Holding the lock during that operation would block all concurrent
				// attempts to try and open a stream, effectively serializing all calls to open the streams.
				numReservedOutputStreams += outputCredit;
				numReservedInputStreams += inputCredit;
			}
			finally {
				lock.unlock();
			}
		}
		catch (InterruptedException e) {
			// restore interruption flag
			Thread.currentThread().interrupt();
			throw new IOException("interrupted before opening stream");
		}

		// open the stream outside the lock.
		boolean success = false;
		try {
			final T out = streamOpener.get();

			// add the stream to the set, need to re-acquire the lock
			lock.lock();
			try {
				openStreams.add(out);
			} finally {
				lock.unlock();
			}

			// good, can now return cleanly
			success = true;
			return out;
		}
		finally {
			if (!success) {
				// remove the reserved credit
				// we must open this non-interruptibly, because this must succeed!
				lock.lock();
				try {
					numReservedOutputStreams -= outputCredit;
					numReservedInputStreams -= inputCredit;
					available.signalAll();
				} finally {
					lock.unlock();
				}
			}
		}
	}

	@GuardedBy("lock")
	private void waitForAvailability(
			int totalLimit,
			int outputLimit,
			int inputLimit) throws InterruptedException, IOException {

		checkState(lock.isHeldByCurrentThread());

		// compute the deadline of this operations
		final long deadline;
		if (streamOpenTimeoutNanos == 0) {
			deadline = Long.MAX_VALUE;
		} else {
			long deadlineNanos = System.nanoTime() + streamOpenTimeoutNanos;
			// check for overflow
			deadline = deadlineNanos > 0 ? deadlineNanos : Long.MAX_VALUE;
		}

		// wait for available connections
		long timeLeft;

		if (streamInactivityTimeoutNanos == 0) {
			// simple case: just wait
			while ((timeLeft = (deadline - System.nanoTime())) > 0 &&
					!hasAvailability(totalLimit, outputLimit, inputLimit)) {

				available.await(timeLeft, TimeUnit.NANOSECONDS);
			}
		}
		else {
			// complex case: chase down inactive streams
			final long checkIntervalNanos = (streamInactivityTimeoutNanos >>> 1) + 1;

			long now;
			while ((timeLeft = (deadline - (now = System.nanoTime()))) > 0 && // while still within timeout
					!hasAvailability(totalLimit, outputLimit, inputLimit)) {

				// check all streams whether there in one that has been inactive for too long
				if (!(closeInactiveStream(openOutputStreams, now) || closeInactiveStream(openInputStreams, now))) {
					// only wait if we did not manage to close any stream.
					// otherwise eagerly check again if we have availability now (we should have!)
					long timeToWait = Math.min(checkIntervalNanos, timeLeft);
					available.await(timeToWait, TimeUnit.NANOSECONDS);
				}
			}
		}

		// check for timeout
		// we check availability again to catch cases where the timeout expired while waiting
		// to re-acquire the lock
		if (timeLeft <= 0 && !hasAvailability(totalLimit, outputLimit, inputLimit)) {
			throw new IOException(String.format(
					"Timeout while waiting for an available stream/connection. " +
					"limits: total=%d, input=%d, output=%d ; Open: input=%d, output=%d ; timeout: %d ms",
					maxNumOpenStreamsTotal, maxNumOpenInputStreams, maxNumOpenOutputStreams,
					numReservedInputStreams, numReservedOutputStreams, getStreamOpenTimeout()));
		}
	}

	@GuardedBy("lock")
	private boolean hasAvailability(int totalLimit, int outputLimit, int inputLimit) {
		return numReservedOutputStreams < outputLimit &&
				numReservedInputStreams < inputLimit &&
				numReservedOutputStreams + numReservedInputStreams < totalLimit;
	}

	@GuardedBy("lock")
	private boolean closeInactiveStream(HashSet<? extends StreamWithTimeout> streams, long nowNanos) {
		for (StreamWithTimeout stream : streams) {
			try {
				final StreamProgressTracker tracker = stream.getProgressTracker();

				// If the stream is closed already, it will be removed anyways, so we
				// do not classify it as inactive. We also skip the check if another check happened too recently.
				if (stream.isClosed() || nowNanos < tracker.getLastCheckTimestampNanos() + streamInactivityTimeoutNanos) {
					// interval since last check not yet over
					return false;
				}
				else if (!tracker.checkNewBytesAndMark(nowNanos)) {
					stream.closeDueToTimeout();
					return true;
				}
			}
			catch (StreamTimeoutException ignored) {
				// may happen due to races
			}
			catch (IOException e) {
				// only log on debug level here, to avoid log spamming
				LOG.debug("Could not check for stream progress to determine inactivity", e);
			}
		}

		return false;
	}

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

	/**
	 * Atomically removes the given output stream from the set of currently open output streams,
	 * and signals that new stream can now be opened.
	 */
	void unregisterOutputStream(OutStream stream) {
		lock.lock();
		try {
			// only decrement if we actually remove the stream
			if (openOutputStreams.remove(stream)) {
				numReservedOutputStreams--;
				available.signalAll();
			}
		}
		finally {
			lock.unlock();
		}
	}

	/**
	 * Atomically removes the given input stream from the set of currently open input streams,
	 * and signals that new stream can now be opened.
	 */
	void unregisterInputStream(InStream stream) {
		lock.lock();
		try {
			// only decrement if we actually remove the stream
			if (openInputStreams.remove(stream)) {
				numReservedInputStreams--;
				available.signalAll();
			}
		}
		finally {
			lock.unlock();
		}
	}

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

	/**
	 * A special IOException, indicating a timeout in the data output stream.
	 */
	public static final class StreamTimeoutException extends IOException {

		private static final long serialVersionUID = -8790922066795901928L;

		public StreamTimeoutException() {
			super("Stream closed due to inactivity timeout. " +
					"This is done to prevent inactive streams from blocking the full " +
					"pool of limited connections");
		}

		public StreamTimeoutException(StreamTimeoutException other) {
			super(other.getMessage(), other);
		}
	}

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

	/**
	 * Interface for streams that can be checked for inactivity.
	 */
	private interface StreamWithTimeout extends Closeable {

		/**
		 * Gets the progress tracker for this stream.
		 */
		StreamProgressTracker getProgressTracker();

		/**
		 * Gets the current position in the stream, as in number of bytes read or written.
		 */
		long getPos() throws IOException;

		/**
		 * Closes the stream asynchronously with a special exception that indicates closing due
		 * to lack of progress.
		 */
		void closeDueToTimeout() throws IOException;

		/**
		 * Checks whether the stream was closed already.
		 */
		boolean isClosed();
	}

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

	/**
	 * A tracker for stream progress. This records the number of bytes read / written together
	 * with a timestamp when the last check happened.
	 */
	private static final class StreamProgressTracker {

		/** The tracked stream. */
		private final StreamWithTimeout stream;

		/** The number of bytes written the last time that the {@link #checkNewBytesAndMark(long)}
		 * method was called. It is important to initialize this with {@code -1} so that the
		 * first check (0 bytes) always appears to have made progress. */
		private volatile long lastCheckBytes = -1;

		/** The timestamp when the last inactivity evaluation was made. */
		private volatile long lastCheckTimestampNanos;

		StreamProgressTracker(StreamWithTimeout stream) {
			this.stream = stream;
		}

		/**
		 * Gets the timestamp when the last inactivity evaluation was made.
		 */
		public long getLastCheckTimestampNanos() {
			return lastCheckTimestampNanos;
		}

		/**
		 * Checks whether there were new bytes since the last time this method was invoked.
		 * This also sets the given timestamp, to be read via {@link #getLastCheckTimestampNanos()}.
		 *
		 * @return True, if there were new bytes, false if not.
		 */
		public boolean checkNewBytesAndMark(long timestamp) throws IOException {
			// remember the time when checked
			lastCheckTimestampNanos = timestamp;

			final long bytesNow = stream.getPos();
			if (bytesNow > lastCheckBytes) {
				lastCheckBytes = bytesNow;
				return true;
			}
			else {
				return false;
			}
		}
	}

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

	/**
	 * A data output stream that wraps a given data output stream and un-registers
	 * from a given connection-limiting file system
	 * (via {@link LimitedConnectionsFileSystem#unregisterOutputStream(OutStream)}
	 * upon closing.
	 */
	private static final class OutStream extends FSDataOutputStream implements StreamWithTimeout {

		/** The original data output stream to write to. */
		private final FSDataOutputStream originalStream;

		/** The connection-limiting file system to un-register from. */
		private final LimitedConnectionsFileSystem fs;

		/** The progress tracker for this stream. */
		private final StreamProgressTracker progressTracker;

		/** An exception with which the stream has been externally closed. */
		private volatile StreamTimeoutException timeoutException;

		/** Flag tracking whether the stream was already closed, for proper inactivity tracking. */
		private AtomicBoolean closed = new AtomicBoolean();

		OutStream(FSDataOutputStream originalStream, LimitedConnectionsFileSystem fs) {
			this.originalStream = checkNotNull(originalStream);
			this.fs = checkNotNull(fs);
			this.progressTracker = new StreamProgressTracker(this);
		}

		// --- FSDataOutputStream API implementation

		@Override
		public void write(int b) throws IOException {
			try {
				originalStream.write(b);
			}
			catch (IOException e) {
				handleIOException(e);
			}
		}

		@Override
		public void write(byte[] b, int off, int len) throws IOException {
			try {
				originalStream.write(b, off, len);
			}
			catch (IOException e) {
				handleIOException(e);
			}
		}

		@Override
		public long getPos() throws IOException {
			try {
				return originalStream.getPos();
			}
			catch (IOException e) {
				handleIOException(e);
				return -1; // silence the compiler
			}
		}

		@Override
		public void flush() throws IOException {
			try {
				originalStream.flush();
			}
			catch (IOException e) {
				handleIOException(e);
			}
		}

		@Override
		public void sync() throws IOException {
			try {
				originalStream.sync();
			}
			catch (IOException e) {
				handleIOException(e);
			}
		}

		@Override
		public void close() throws IOException {
			if (closed.compareAndSet(false, true)) {
				try {
					originalStream.close();
				}
				catch (IOException e) {
					handleIOException(e);
				}
				finally {
					fs.unregisterOutputStream(this);
				}
			}
		}

		@Override
		public void closeDueToTimeout() throws IOException {
			this.timeoutException = new StreamTimeoutException();
			close();
		}

		@Override
		public boolean isClosed() {
			return closed.get();
		}

		@Override
		public StreamProgressTracker getProgressTracker() {
			return progressTracker;
		}

		private void handleIOException(IOException exception) throws IOException {
			if (timeoutException == null) {
				throw exception;
			} else {
				// throw a new exception to capture this call's stack trace
				// the new exception is forwarded as a suppressed exception
				StreamTimeoutException te = new StreamTimeoutException(timeoutException);
				te.addSuppressed(exception);
				throw te;
			}
		}
	}

	/**
	 * A data input stream that wraps a given data input stream and un-registers
	 * from a given connection-limiting file system
	 * (via {@link LimitedConnectionsFileSystem#unregisterInputStream(InStream)}
	 * upon closing.
	 */
	private static final class InStream extends FSDataInputStream implements StreamWithTimeout {

		/** The original data input stream to read from. */
		private final FSDataInputStream originalStream;

		/** The connection-limiting file system to un-register from. */
		private final LimitedConnectionsFileSystem fs;

		/** An exception with which the stream has been externally closed. */
		private volatile StreamTimeoutException timeoutException;

		/** The progress tracker for this stream. */
		private final StreamProgressTracker progressTracker;

		/** Flag tracking whether the stream was already closed, for proper inactivity tracking. */
		private AtomicBoolean closed = new AtomicBoolean();

		InStream(FSDataInputStream originalStream, LimitedConnectionsFileSystem fs) {
			this.originalStream = checkNotNull(originalStream);
			this.fs = checkNotNull(fs);
			this.progressTracker = new StreamProgressTracker(this);
		}

		// --- FSDataOutputStream API implementation

		@Override
		public int read() throws IOException {
			try {
				return originalStream.read();
			}
			catch (IOException e) {
				handleIOException(e);
				return 0; // silence the compiler
			}
		}

		@Override
		public int read(byte[] b) throws IOException {
			try {
				return originalStream.read(b);
			}
			catch (IOException e) {
				handleIOException(e);
				return 0; // silence the compiler
			}
		}

		@Override
		public int read(byte[] b, int off, int len) throws IOException {
			try {
				return originalStream.read(b, off, len);
			}
			catch (IOException e) {
				handleIOException(e);
				return 0; // silence the compiler
			}
		}

		@Override
		public long skip(long n) throws IOException {
			try {
				return originalStream.skip(n);
			}
			catch (IOException e) {
				handleIOException(e);
				return 0L; // silence the compiler
			}
		}

		@Override
		public int available() throws IOException {
			try {
				return originalStream.available();
			}
			catch (IOException e) {
				handleIOException(e);
				return 0; // silence the compiler
			}
		}

		@Override
		public void mark(int readlimit) {
			originalStream.mark(readlimit);
		}

		@Override
		public void reset() throws IOException {
			try {
				originalStream.reset();
			}
			catch (IOException e) {
				handleIOException(e);
			}
		}

		@Override
		public boolean markSupported() {
			return originalStream.markSupported();
		}

		@Override
		public void seek(long desired) throws IOException {
			try {
				originalStream.seek(desired);
			}
			catch (IOException e) {
				handleIOException(e);
			}
		}

		@Override
		public long getPos() throws IOException {
			try {
				return originalStream.getPos();
			}
			catch (IOException e) {
				handleIOException(e);
				return 0; // silence the compiler
			}
		}

		@Override
		public void close() throws IOException {
			if (closed.compareAndSet(false, true)) {
				try {
					originalStream.close();
				}
				catch (IOException e) {
					handleIOException(e);
				}
				finally {
					fs.unregisterInputStream(this);
				}
			}
		}

		@Override
		public void closeDueToTimeout() throws IOException {
			this.timeoutException = new StreamTimeoutException();
			close();
		}

		@Override
		public boolean isClosed() {
			return closed.get();
		}

		@Override
		public StreamProgressTracker getProgressTracker() {
			return progressTracker;
		}

		private void handleIOException(IOException exception) throws IOException {
			if (timeoutException == null) {
				throw exception;
			} else {
				// throw a new exception to capture this call's stack trace
				// the new exception is forwarded as a suppressed exception
				StreamTimeoutException te = new StreamTimeoutException(timeoutException);
				te.addSuppressed(exception);
				throw te;
			}
		}
	}

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

	/**
	 * A simple configuration data object capturing the settings for limited connections.
	 */
	public static class ConnectionLimitingSettings {

		/** The limit for the total number of connections, or 0, if no limit. */
		public final int limitTotal;

		/** The limit for the number of input stream connections, or 0, if no limit. */
		public final int limitInput;

		/** The limit for the number of output stream connections, or 0, if no limit. */
		public final int limitOutput;

		/** The stream opening timeout for a stream, in milliseconds. */
		public final long streamOpenTimeout;

		/** The inactivity timeout for a stream, in milliseconds. */
		public final long streamInactivityTimeout;

		/**
		 * Creates a new ConnectionLimitingSettings with the given parameters.
		 *
		 * @param limitTotal The limit for the total number of connections, or 0, if no limit.
		 * @param limitInput The limit for the number of input stream connections, or 0, if no limit.
		 * @param limitOutput The limit for the number of output stream connections, or 0, if no limit.
		 * @param streamOpenTimeout       The maximum number of milliseconds that the file system will wait when
		 *                                no more connections are currently permitted.
		 * @param streamInactivityTimeout The milliseconds that a stream may spend not writing any
		 *                                bytes before it is closed as inactive.
		 */
		public ConnectionLimitingSettings(
				int limitTotal,
				int limitInput,
				int limitOutput,
				long streamOpenTimeout,
				long streamInactivityTimeout) {
			checkArgument(limitTotal >= 0);
			checkArgument(limitInput >= 0);
			checkArgument(limitOutput >= 0);
			checkArgument(streamOpenTimeout >= 0);
			checkArgument(streamInactivityTimeout >= 0);

			this.limitTotal = limitTotal;
			this.limitInput = limitInput;
			this.limitOutput = limitOutput;
			this.streamOpenTimeout = streamOpenTimeout;
			this.streamInactivityTimeout = streamInactivityTimeout;
		}

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

		/**
		 * Parses and returns the settings for connection limiting, for the file system with
		 * the given file system scheme.
		 *
		 * @param config The configuration to check.
		 * @param fsScheme The file system scheme.
		 *
		 * @return The parsed configuration, or null, if no connection limiting is configured.
		 */
		@Nullable
		public static ConnectionLimitingSettings fromConfig(Configuration config, String fsScheme) {
			checkNotNull(fsScheme, "fsScheme");
			checkNotNull(config, "config");

			final ConfigOption<Integer> totalLimitOption = CoreOptions.fileSystemConnectionLimit(fsScheme);
			final ConfigOption<Integer> limitInOption = CoreOptions.fileSystemConnectionLimitIn(fsScheme);
			final ConfigOption<Integer> limitOutOption = CoreOptions.fileSystemConnectionLimitOut(fsScheme);

			final int totalLimit = config.getInteger(totalLimitOption);
			final int limitIn = config.getInteger(limitInOption);
			final int limitOut = config.getInteger(limitOutOption);

			checkLimit(totalLimit, totalLimitOption);
			checkLimit(limitIn, limitInOption);
			checkLimit(limitOut, limitOutOption);

			// create the settings only, if at least one limit is configured
			if (totalLimit <= 0 || limitIn <= 0 || limitOut <= 0) {
				// no limit configured
				return null;
			}
			else {
				final ConfigOption<Long> openTimeoutOption =
						CoreOptions.fileSystemConnectionLimitTimeout(fsScheme);
				final ConfigOption<Long> inactivityTimeoutOption =
						CoreOptions.fileSystemConnectionLimitStreamInactivityTimeout(fsScheme);

				final long openTimeout = config.getLong(openTimeoutOption);
				final long inactivityTimeout = config.getLong(inactivityTimeoutOption);

				checkTimeout(openTimeout, openTimeoutOption);
				checkTimeout(inactivityTimeout, inactivityTimeoutOption);

				return new ConnectionLimitingSettings(
						totalLimit == -1 ? 0 : totalLimit,
						limitIn == -1 ? 0 : limitIn,
						limitOut == -1 ? 0 : limitOut,
						openTimeout,
						inactivityTimeout);
			}
		}

		private static void checkLimit(int value, ConfigOption<Integer> option) {
			if (value < -1) {
				throw new IllegalConfigurationException("Invalid value for '" + option.key() + "': " + value);
			}
		}

		private static void checkTimeout(long timeout, ConfigOption<Long> option) {
			if (timeout < 0) {
				throw new IllegalConfigurationException("Invalid value for '" + option.key() + "': " + timeout);
			}
		}
	}
}


text:
package org.eclipse.rdf4j.sail.solr;

import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.sail.lucene.DocumentDistance;
import org.eclipse.rdf4j.sail.lucene.util.GeoUnits;

public class SolrDocumentDistance extends SolrDocumentResult implements DocumentDistance {

	private final IRI units;

	public SolrDocumentDistance(SolrSearchDocument doc, IRI units) {
		super(doc);
		this.units = units;
	}

	@Override
	public double getDistance() {
		Number s = ((Number) doc.getDocument().get(SolrIndex.DISTANCE_FIELD));
		return (s != null) ? GeoUnits.fromKilometres(s.doubleValue(), units) : Double.NaN;
	}
}

redpajama_set_name: RedPajamaGithub


text:
//
//  ComponentInstance.hpp
//
// Sketchup C++ Wrapper for C API
// MIT License
//
// Copyright (c) 2017 Tom Kaneko
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:

// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.

// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//

#ifndef ComponentInstance_hpp
#define ComponentInstance_hpp

#include <stdio.h>

#include <SketchUpAPI/model/component_instance.h>

#include "SUAPI-CppWrapper/Transformation.hpp"
#include "SUAPI-CppWrapper/model/Entity.hpp"
#include "SUAPI-CppWrapper/model/ComponentDefinition.hpp"

namespace CW {

class ComponentInstance :public DrawingElement {
  friend class Group;
  protected:
  /** Creates a copy of the ComponentInstance */
  SUComponentInstanceRef copy_reference(const ComponentInstance& other);

  public:
  /**
  * Constructor for null object
  */
  ComponentInstance();

  ComponentInstance(SUComponentInstanceRef instance, bool attached = true);

  /** Copy constructor */
  ComponentInstance(const ComponentInstance& other);

  /** Copy constructor for derived classes (Group) */
  ComponentInstance(const ComponentInstance& other, SUComponentInstanceRef instance_ref);

  /** Destructor */
  ~ComponentInstance();

  /** Copy assignment operator */
  ComponentInstance& operator=(const ComponentInstance& other);

  /**
  * Returns the raw SUComponentInstance object.
  */
  SUComponentInstanceRef ref() const;

  /** Cast to native objects */
  operator SUComponentInstanceRef() const;
  operator SUComponentInstanceRef*();

  /**
  * Returns the Transformation object applied to this instance.
  */
  Transformation transformation() const;

  /**
  * Set the transformation applied to this Component Instance
  */
  void transformation(const Transformation& tranform);

  /**
  * Returns the Component Definition object of this instance.
  */
  ComponentDefinition definition() const;

  /**
  * Returns the name of this instance.
  */
  String name() const;

  /**
  * Sets the name of this instance.
  */
  void name(const String& string);

};

} /* namespace CW */
#endif /* ComponentInstance_hpp */


text:
package org.kohsuke.args4j.spi;

import java.util.List;

/**
 * Interface that can be instantiated to get default values.
 * @see Setter
 * @author Stephan Fuhrmann
 */
public interface Getter<T> {
    /**
     * Gets the current value of the property.
     *
     * <p>
     * A {@link Getter} object has an implicit knowledge about the property it's getting,
     * and the instance of the option bean.
     *
     * @return
     *      empty list or null if there's no current value.
     */
    List<T> getValueList();
}


text:
package com.svoflee.spartacus.core.domain;

import java.io.Serializable;
import java.util.Date;

import javax.persistence.Column;
import javax.persistence.MappedSuperclass;
import javax.persistence.Temporal;
import javax.persistence.TemporalType;

import com.google.code.morphia.annotations.Property;
import com.svoflee.spartacus.core.ui.annotation.Editable;
import com.svoflee.spartacus.core.ui.annotation.ShowType;
import com.svoflee.spartacus.core.ui.annotation.UI;
import com.svoflee.spartacus.core.ui.model.FieldType;

/**
 * BaseAuditableEntity 是默认的通用的Domain Entity定义,
 * 它包含了通用的属性定义,包括数据审计信息及默认扩展码
 *
 * @author svoflee
 * @since 1.0.0
 * @version 1.0.0
 * @param <PK>
 */
@MappedSuperclass
public abstract class BaseAuditableEntity<PK extends Serializable> extends BasePersistableEntity<PK> implements
        DataAuditable {

    // public static final String STATUS_PROPERTY_NAME = "status";

    // public static final String IS_AVAILABLE_PROPERTY_NAME = "isAvailable";

    private static final long serialVersionUID = -5134920365779298970L;

    public static final String DEFAULT_OPERATOR = "SYS";

    @UI(caption = "创建时间", index = 9911, editable = Editable.NONE, fieldType = FieldType.DATETIME, showType = ShowType.FORM)
    @Property("ct")
    @Column(name = "CREATED_TIME", nullable = false)
    @Temporal(TemporalType.TIMESTAMP)
    private Date createdTime = new Date();

    @UI(caption = "创建者", index = 9920, editable = Editable.NONE, fieldType = FieldType.STRING, showType = ShowType.FORM)
    @Property("cb")
    @Column(name = "CREATED_BY", nullable = false, length = 20)
    private String createdBy = BaseAuditableEntity.DEFAULT_OPERATOR;

    @UI(caption = "更新时间", index = 9910, editable = Editable.NONE, fieldType = FieldType.DATETIME, showType = ShowType.FORM)
    @Property("ut")
    @Column(name = "UPDATED_TIME", nullable = false)
    @Temporal(TemporalType.TIMESTAMP)
    private Date updatedTime = new Date();

    @UI(caption = "更新者", index = 9920, editable = Editable.NONE, fieldType = FieldType.STRING, showType = ShowType.FORM)
    @Property("ub")
    @Column(name = "UPDATED_BY", nullable = false, length = 20)
    private String updatedBy = BaseAuditableEntity.DEFAULT_OPERATOR;

    // @UI(caption = "版本号", index = 9830, editable = Editable.NONE, fieldType = FieldType.NUMBER, showType =
    // ShowType.NO)
    // @Version("m_v")
    // @Column(name = "M_VERSION", nullable = false)
    // private Long version = 1L;
    // public void setVersion(Long version) {
    // this.version = version;
    // }
    //
    // public Long getVersion() {
    // return this.version;
    // }

    @Override
    public void setCreatedTime(Date createdTime) {
        this.createdTime = createdTime;
    }

    @Override
    public Date getCreatedTime() {
        return this.createdTime;
    }

    @Override
    public void setCreatedBy(String createdBy) {
        this.createdBy = createdBy;
    }

    @Override
    public String getCreatedBy() {
        return this.createdBy;
    }

    @Override
    public void setUpdatedBy(String updatedBy) {
        this.updatedBy = updatedBy;
    }

    @Override
    public String getUpdatedBy() {
        return this.updatedBy;
    }

    @Override
    public void setUpdatedTime(Date updatedTime) {
        this.updatedTime = updatedTime;
    }

    @Override
    public Date getUpdatedTime() {
        return this.updatedTime;
    }

}


text:
from .base import Visitor
from ..result import ReleaseResult

class SerializeVisitor(Visitor):

    ARTIST_TYPE_TRANSLATION = {
        ReleaseResult.ArtistTypes.MAIN: "ArtistTypes.MAIN",
        ReleaseResult.ArtistTypes.FEATURING: "ArtistTypes.FEATURING",
        ReleaseResult.ArtistTypes.REMIXER: "ArtistTypes.REMIXER"
    }

    ALBUM_ART_TYPE_TRANSLATION = {
        ReleaseResult.AlbumArtTypes.BACK: "AlbumArtTypes.BACK",
        ReleaseResult.AlbumArtTypes.DISC: "AlbumArtTypes.DISC",
        ReleaseResult.AlbumArtTypes.FRONT: "AlbumArtTypes.FRONT",
        ReleaseResult.AlbumArtTypes.INLET: "AlbumArtTypes.INLET"
    }

    def __init__(self, var_name):
        self.var_name = var_name

    def _serialize_artists(self, artists, var_name, track_var_name=None, empty_line=True):
        code = []
        for artist in artists:
            code.append(u'%s = %s.create_artist()' % (var_name, self.var_name))
            code.append(u'%s.set_name(%r)' % (var_name, artist.get_name()))
            code.append(u'%s.set_various(%r)' % (var_name, artist.is_various()))
            for artist_type in artist.get_types():
                code.append(u'%s.append_type(%s.%s)' % (var_name, self.var_name, self.ARTIST_TYPE_TRANSLATION[artist_type]))
            if track_var_name:
                code.append(u'%s.append_artist(%s)' % (track_var_name, var_name))
            else:
                code.append(u'%s.append_release_artist(%s)' % (self.var_name, var_name))
            if empty_line:
                code.append(u'')
        return code

    def visit_NotFoundResult(self, result):
        code = []
        code.append(u'%s = NotFoundResult()' % self.var_name)
        code.append(u'%s.set_scraper_name(%r)' % (self.var_name, result.get_scraper_name()))
        return u'\n'.join(code)

    def visit_ListResult(self, result):
        code = []
        code.append(u'%s = ListResult()' % self.var_name)
        code.append(u'%s.set_scraper_name(%r)' % (self.var_name, result.get_scraper_name()))
        for item in result.get_items():
            code.append('')
            code.append(u'item = %s.create_item()' % self.var_name)
            code.append(u'item.set_name(%r)' % item.get_name())
            code.append(u'item.set_info(%r)' % item.get_info())
            code.append(u'item.set_query(%r)' % item.get_query())
            code.append(u'item.set_query_scraper(%r)' % item.get_query_scraper())
            code.append(u'item.set_url(%r)' % item.get_url())
            code.append(u'%s.append_item(item)' % self.var_name)
        return u'\n'.join(code)

    def visit_ReleaseResult(self, result):
        code = []
        code.append(u'%s = ReleaseResult()' % self.var_name)
        code.append(u'%s.set_scraper_name(%r)' % (self.var_name, result.get_scraper_name()))
        code.append(u'')
        for release_event in result.get_release_events():
            code.append(u'release_event = %s.create_release_event()' % self.var_name)
            code.append(u'release_event.set_date(%r)' % release_event.get_date())
            code.append(u'release_event.set_country(%r)' % release_event.get_country())
            code.append(u'%s.append_release_event(release_event)' % self.var_name)
            code.append(u'')
        code.append(u'%s.set_format(%r)' % (self.var_name, result.get_format()))
        code.append(u'')
        for label_id in result.get_label_ids():
            code.append(u'label_id = %s.create_label_id()' % self.var_name)
            code.append(u'label_id.set_label(%r)' % label_id.get_label())
            for catalogue_nr in label_id.get_catalogue_nrs():
                code.append(u'label_id.append_catalogue_nr(%r)' % catalogue_nr)
            code.append(u'%s.append_label_id(label_id)' % self.var_name)
            code.append(u'')
        code.append(u'%s.set_title(%r)' % (self.var_name, result.get_title()))
        code.append(u'')
        code.extend(self._serialize_artists(result.get_release_artists(), u'artist'))
        for genre in result.get_genres():
            code.append(u'%s.append_genre(%r)' % (self.var_name, genre))
        code.append(u'')
        for style in result.get_styles():
            code.append(u'%s.append_style(%r)' % (self.var_name, style))
        code.append(u'')
        code.append(u'%s.set_url(%r)' % (self.var_name, result.get_url()))
        code.append(u'')
        for disc in result.get_discs():
            code.append(u'disc = %s.create_disc()' % self.var_name)
            code.append(u'disc.set_number(%r)' % disc.get_number())
            code.append(u'disc.set_title(%r)' % disc.get_title())
            code.append(u'')
            for track in disc.get_tracks():
                code.append(u'track = disc.create_track()')
                code.append(u'track.set_number(%r)' % track.get_number())
                code.append(u'track.set_title(%r)' % track.get_title())
                code.append(u'track.set_length(%r)' % track.get_length())
                code.extend(self._serialize_artists(track.get_artists(), u'track_artist', u'track', False))
                code.append(u'disc.append_track(track)')
                code.append(u'')
            code.append(u'%s.append_disc(disc)' % self.var_name)
            code.append(u'')
        for album_art in result.get_album_arts():
            code.append(u'album_art = %s.create_album_art()' % self.var_name)
            code.append(u'album_art.set_url(%r)' % album_art.get_url())
            code.append(u'album_art.set_type(%s.%s)' % (self.var_name, self.ALBUM_ART_TYPE_TRANSLATION[album_art.get_type()]))
            code.append(u'album_art.set_width(%r)' % album_art.get_width())
            code.append(u'album_art.set_height(%r)' % album_art.get_height())
            code.append(u'album_art.set_thumbnail_url(%r)' % album_art.get_thumbnail_url())
            code.append(u'album_art.set_thumbnail_width(%r)' % album_art.get_thumbnail_width())
            code.append(u'album_art.set_thumbnail_height(%r)' % album_art.get_thumbnail_height())
            code.append(u'album_art.set_hint(%r)' % album_art.geT_hint())
            code.append(u'%s.append_album_art(album_art)' % self.var_name)
            code.append(u'')
        return u'\n'.join(code)

class JSONSerializeVisitor(Visitor):

    RESULT_TYPE_NAME = u'type'

    ALBUM_ART_TYPE_TRANSLATION = {
        ReleaseResult.AlbumArtTypes.BACK: u'back',
        ReleaseResult.AlbumArtTypes.DISC: u'disc',
        ReleaseResult.AlbumArtTypes.FRONT: u'front',
        ReleaseResult.AlbumArtTypes.INLET: u'inlet'
    }

    ARTIST_TYPE_TRANSLATION = {
        ReleaseResult.ArtistTypes.MAIN: u'main',
        ReleaseResult.ArtistTypes.FEATURING: u'guest',
        ReleaseResult.ArtistTypes.REMIXER: u'remixer'
    }

    def _unicode_or_none(self, str):
        if str is not None:
            return unicode(str)
        else:
            return None

    def visit_NotFoundResult(self, result):
        out = {
            self.RESULT_TYPE_NAME: u'NotFoundResult'
        }
        return out

    def visit_ListItem(self, item):
        out = {
            u'name':          self._unicode_or_none(item.get_name()),
            u'info':          self._unicode_or_none(item.get_info()),
            u'query':         self._unicode_or_none(item.get_query()),
            u'query_scraper': self._unicode_or_none(item.get_query_scraper()),
            u'url':           self._unicode_or_none(item.get_url())
        }
        return out

    def visit_ListResult(self, result):
        out = {
            self.RESULT_TYPE_NAME: u'ListResult',

            u'items': map(lambda x: self.visit(x), result.get_items())
        }
        return out

    def visit_AlbumArt(self, albumart):
        out = {
            u'url':    self._unicode_or_none(albumart.get_url()),
            u'type':   self.ALBUM_ART_TYPE_TRANSLATION[albumart.get_type()] if albumart.get_type() else None,
            u'width':  albumart.get_width(),
            u'height': albumart.get_height(),
            u'hint':   self._unicode_or_none(albumart.get_hint())
        }
        return out

    def visit_ReleaseEvent(self, rel_event):
        out = {
            u'date':    self._unicode_or_none(rel_event.get_date()),
            u'country': self._unicode_or_none(rel_event.get_country())
        }
        return out

    def visit_LabelId(self, labelId):
        out = {
            u'label':        self._unicode_or_none(labelId.get_label()),
            u'catalogueNrs': map(lambda x: self._unicode_or_none(x), labelId.get_catalogue_nrs())
        }
        return out

    def visit_Artist(self, artist):
        out = {
            u'name':      self._unicode_or_none(artist.get_name()),
            u'types':     map(lambda x: self.ARTIST_TYPE_TRANSLATION[x], artist.get_types()),
            u'isVarious': artist.is_various()
        }
        return out

    def visit_Track(self, track):
        out = {
            u'number':  self._unicode_or_none(track.get_number()),
            u'artists': map(lambda x: self.visit(x), track.get_artists()),
            u'title':   self._unicode_or_none(track.get_title()),
            u'length':  track.get_length()
        }
        return out

    def visit_Disc(self, disc):
        out = {
            u'number': disc.get_number(),
            u'title':  disc.get_title(),
            u'tracks': map(lambda x: self.visit(x), disc.get_tracks())
        }
        return out

    def visit_ReleaseResult(self, result):
        out = {
            self.RESULT_TYPE_NAME: u'ReleaseResult',

            u'releaseEvents': map(lambda x: self.visit(x), result.get_release_events()),
            u'format':        self._unicode_or_none(result.get_format()),
            u'labelIds':      map(lambda x: self.visit(x), result.get_label_ids()),
            u'title':         self._unicode_or_none(result.get_title()),
            u'artists':       map(lambda x: self.visit(x), result.get_release_artists()),
            u'genres':        map(lambda x: self._unicode_or_none(x), result.get_genres()),
            u'styles':        map(lambda x: self._unicode_or_none(x), result.get_styles()),
            u'url':           self._unicode_or_none(result.get_url()),
            u'discs':         map(lambda x: self.visit(x), result.get_discs()),
            u'albumArts':     map(lambda x: self.visit(x), result.get_album_arts())
        }
        return out


text:
module AUTOSAR.Shared.PID
  ( -- * Full PID controller
    PIDCtrl(..)
  , pidController
    -- * PD controller
  , pdController
    -- * PI controller
  , piController
  ) where

import AUTOSAR.ARSim

-- * PID controller
-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
-- The PID controller produces a linear combination of Propotional-,
-- Integrating and Differentiating gain. A few useful special cases (PI-, PD-)
-- are exported as well.

-- | The PID controller requires a @(cruise, vehicle)@ velocity tuple and
-- provides a throttle control.
newtype PIDCtrl = PIDCtrl
  { pidOp :: ClientServerOp (Double, Double) Double Provided }

-- | Create a PID controller.
pidController :: Time
              -- ^ Sample time
              -> Time
              -- ^ Derivative time
              -> Time
              -- ^ Integral time
              -> Double
              -- ^ Proportional scale
              -> AUTOSAR PIDCtrl
              -- ^ Throttle control (output)
pidController dt td ti k = atomic $ do
     state <- interRunnableVariable (0.0, 0.0)
     pidOp <- providedPort
     serverRunnable (MinInterval 0) [OperationInvokedEvent pidOp] $
       \(ctrl, feedback) -> do
          Ok (prevErr, prevInt) <- rteIrvRead state

          let err        = ctrl - feedback
              derivative = (err - prevErr) / dt
              integral   = prevInt + dt / ti * err
              output     = k * (err + integral + td * derivative)
          rteIrvWrite state (err, integral)
          return output
     return $ sealBy PIDCtrl pidOp

-- | Create a PD controller (proportional and derivative control).
pdController :: Time
             -- ^ Sample time
             -> Time
             -- ^ Derivative time
             -> Double
             -- ^ Proportional scale
             -> AUTOSAR PIDCtrl
pdController deltaT td = pidController deltaT td (1/0)

-- | Create a PI controller (proportional and integral control).
piController :: Time
             -- ^ Sample time
             -> Time
             -- ^ Integral time
             -> Double
             -- ^ Proportional scale
             -> AUTOSAR PIDCtrl
piController deltaT = pidController deltaT 0


text:
/**
 * Main public API classes of the core streaming JSON
 * processor: most importantly {@link com.fasterxml.jackson.core.JsonFactory}
 * used for constructing
 * JSON parser ({@link com.fasterxml.jackson.core.JsonParser})
 * and generator
 * ({@link com.fasterxml.jackson.core.JsonGenerator})
 * instances.
 *

 * Public API of the higher-level mapping interfaces ("Mapping API")
 * is found from the "jackson-databind" bundle, except for following
 * base interfaces that are defined here:
 *

 *
{@link com.fasterxml.jackson.core.TreeNode} is included
 *within Streaming API to support integration of the Tree Model
 *(which is based on JsonNode) with the basic
 *parsers and generators (iff using mapping-supporting factory: which
 *is part of Mapping API, not core)
 *

 *
{@link com.fasterxml.jackson.core.ObjectCodec} is included so that
 *  reference to the object capable of serializing/deserializing
 *  Objects to/from JSON (usually, com.fasterxml.jackson.databind.ObjectMapper)
 *  can be exposed, without adding direct dependency to implementation.
 *

 *

 */

package com.fasterxml.jackson.core;


text:
<?php

	require_once(TOOLKIT . '/class.event.php');

	Class eventsave_milestone extends Event{

		const ROOTELEMENT = 'save-milestone';

		public $eParamFILTERS = array(

		);

		public static function about(){
			return array(
					 'name' => 'Save Milestone',
					 'author' => array(
							'name' => 'Stephen Bau',
							'website' => 'http://home/sym/form-builder',
							'email' => 'bauhouse@gmail.com'),
					 'version' => '1.0',
					 'release-date' => '2010-03-15T06:58:01+00:00',
					 'trigger-condition' => 'action[save-milestone]');
		}

		public static function getSource(){
			return '3';
		}

		public static function allowEditorToParse(){
			return true;
		}

		public static function documentation(){
			return '

Success and Failure XML Examples

When saved successfully, the following XML will be returned:

<save-milestone result="success" type="create | edit">
  <message>Entry [created | edited] successfully.</message>
</save-milestone>

When an error occurs during saving, due to either missing or invalid fields, the following XML will be returned:

<save-milestone result="error">
  <message>Entry encountered errors when saving.</message>
  <field-name type="invalid | missing" />
  ...
</save-milestone>

Example Front-end Form Markup

This is an example of the form markup you can use on your frontend:

<form method="post" action="" enctype="multipart/form-data">
  <input name="MAX_FILE_SIZE" type="hidden" value="5242880" />
  <label>Title
    <input name="fields[title]" type="text" />
  </label>
  <label>Description
    <textarea name="fields[description]" rows="15" cols="50"></textarea>
  </label>
  <label>Status
    <select name="fields[status]">
      <option value="Active">Active</option>
      <option value="Inactive">Inactive</option>
      <option value="Finished">Finished</option>
    </select>
  </label>
  <input name="fields[project]" type="hidden" value="..." />
  <label>Due
    <input name="fields[due]" type="text" />
  </label>
  <input name="action[save-milestone]" type="submit" value="Submit" />
</form>

To edit an existing entry, include the entry ID value of the entry in the form. This is best as a hidden field like so:

<input name="id" type="hidden" value="23" />

To redirect to a different location upon a successful save, include the redirect location in the form. This is best as a hidden field like so, where the value is the URL to redirect to:

<input name="redirect" type="hidden" value="http://home/sym/form-builder/success/" />
';
		}

		public function load(){
			if(isset($_POST['action']['save-milestone'])) return $this->__trigger();
		}

		protected function __trigger(){
			include(TOOLKIT . '/events/event.section.php');
			return $result;
		}

	}


text:
<TITLE>Review for Broken English (1996)</TITLE>
<LINK REL="STYLESHEET" TYPE="text/css" HREF="/ramr.css">

Broken English (1996)

reviewed by
James Berardinelli

                                   BROKEN ENGLISH
                       A film review by James Berardinelli
                        Copyright 1997 James Berardinelli

RATING (0 TO 10): 6.0
Alternative Scale: **1/2 out of ****

New Zealand, 1996
U.S. Release Date: 5/2/97 (limited)
Running Length: 1:30
MPAA Classification: R (Profanity, sex, violence)
Theatrical Aspect Ratio: 1.85:1

Cast: Aleksandra Vujcic, Julian Arahanga, Rade Serbedzija,
      Marton Csokas, Madeline McNamara, Elizabeth Mavric, Jing Zhao,
      Li Yang, Temuera Morrison
Director: Gregor Nicholas
Producer: Rogin Scholes
Screenplay: Gregor Nicholas, Johanna Pigott, Jim Salter
Cinematography: John Toon
Music: Murray Grindlay and Murray McNabb
U.S. Distributor: Sony Pictures Classics

     BROKEN ENGLISH deals with clashing cultures in New Zealand, "the
bottom of the world."  There's an energy in this film, the first from
director Gregor Nicholas, that demands to be noticed, but it's not
enough to overcome the weaknesses of a predictably shallow, derivative
script.  The only new ingredient in BROKEN ENGLISH's approach to racial
intolerance and strife is that the tension is between immigrant
Croatians and native Maoris.  The basic story, however, is familiar, and
the characters aren't sufficiently well-developed to captivate an
audience.

     A secondary aspect of BROKEN ENGLISH addresses a father's obsessive
need to control the lives of his children.  Like SHINE's Peter Helfgott,
Ivan (Rade Serbedzija) is determined to keep his family together at all
costs, and is willing to use violence to thwart any non-Croatian who
shows an interest in either of his daughters.  However, where Armin
Mueller-Stahl was able to fashion a multi-dimensional portrait of a torn
man, Ivan is a distressingly flat villain.  Early in the film, there are
shades to his personality, but, as the story progresses, he becomes
progressively more of a one-note character, and any grudging sympathy we
might have for him is destroyed by the time we reach the overly
melodramatic climax.

     The premise is not inherently unpromising, despite (or perhaps
because of) certain similarities to that of ONCE WERE WARRIORS.
Accompanied by his immediate family, Ivan has fled the battlefield of
his native country, Croatia, to settle in Auckland.   There, he earns
his living by growing and distributing marijuana.  His youngest
daughter, Nina (Aleksandra Vujcic), falls in love with Eddie (Julian
Arahanga), a Maori cook who works at the restaurant where she's a
waitress.  Eventually, Nina moves away from home, but Ivan's disapproval
of Eddie isn't the only reason.  In order to help a Chinese friend gain
residency status, Nina (who is a citizen of New Zealand because her
mother was born there) agrees to a marriage in name only.  Ultimately,
Nina wants to integrate all aspects of her life into one cohesive whole,
but her attempts to reconcile with Ivan lead to violence and grief.

     Nicholas doesn't offer anything revolutionary in his approach to
warring cultures, and the story of two lovers defying tradition and
family wishes is older than ROMEO AND JULIET.  In fact, the basic themes
and framework of BROKEN ENGLISH have been explored more effectively in
films like MISSISSIPPI MASALA and DOUBLE HAPPINESS.  Racial purity and
isolation are the issues here.  Wanting his grandchildren to be full-
blooded Croatians,  Ivan despises the idea of interracial breeding.
This is the kind of narrow-minded bigotry that begins race wars and
leads to attempted genocides.

     The main actors turn in solid performances.  Veteran Rade
Serbedzija, best-known to American audiences from BEFORE THE RAIN, is
terrific at portraying rage, but somewhat less successful at conveying
pain and loss.  The sensual Aleksandra Vujcic is surprisingly effective
for a non-professional in her first role.  Julian Arahanga smolders as
Eddie, and while he and Vujcic don't have perfect chemistry, they steam
up the screen during one particularly hot sex scene.  Some of the
supporting players (Marton Csokas, as Nina's brother, in particular)
aren't impressive.  Temuera Morrison (the male lead from ONCE WERE
WARRIORS) has a small part as Eddie's brother.

     BROKEN ENGLISH -- the title refers to the common language that
immigrants must speak in order to communicate -- was filmed with a great
deal of energy and color.  As absurd as the climax is, it's still
peculiarly compelling, if only because there's so much raw violence and
hatred percolating on-screen.  Unfortunately, the characters don't hold
the same fascination as the visual and visceral elements, and, as a
result, the trappings of BROKEN ENGLISH are almost more interesting than
the actual story.  So, instead of being an equal to ONCE WERE WARRIORS,
BROKEN ENGLISH is a faint, fleeting shadow.

- James Berardinelli
e-mail: berardin@bc.cybernex.net
ReelViews web site: http://www.cybernex.net/~berardin

The review above was posted to the
rec.arts.movies.reviews newsgroup (de.rec.film.kritiken for German reviews).

The Internet Movie Database accepts no responsibility for the contents of the
review and has no editorial control. Unless stated otherwise, the copyright
belongs to the author.

Please direct comments/criticisms of the review to relevant newsgroups.

Broken URLs inthe reviews are the responsibility of the author.

The formatting of the review is likely to differ from the original due
to ASCII to HTML conversion.

Related links: index of all rec.arts.movies.reviews reviews


text:
NS_ASSUME_NONNULL_BEGIN

@interface CDABGDownloadingFile : NSManagedObject

// Insert code here to declare functionality of your managed object subclass

@end

NS_ASSUME_NONNULL_END

#import "CDABGDownloadingFile+CoreDataProperties.h"


text: obj-$(CONFIG_REED_SOLOMON) += reed_solomon.o


text:
class AddFormatToDatapackageResourceField < ActiveRecord::Migration
  def change
    add_column :datapackage_resource_fields, :format, :text
  end
end


text:
FactoryGirl.define do
  factory :outfit do
    image_url 'http://lorempixel.com/400/200/people'
    association :user
    title { Faker::Name.title}
    gender 'Ladies'

    trait :without_hashtags do
      caption {Faker::Hacker.say_something_smart}
    end

    trait :with_hashtags do
      caption {Faker::Hacker.say_something_smart + " #" + Faker::Hacker.ingverb }
    end
  end
end


text:
var Server = require('bittorrent-tracker').Server;

var server = new Server({
  udp: true, // enable udp server? [default=true]
  http: true, // enable http server? [default=true]
  ws: true, // enable websocket server? [default=true]
  stats: true // enable web-based statistics? [default=true]
});

server.on('error', function (err) {
  // fatal server error!
  console.log(err.message);
});

server.on('warning', function (err) {
  // client sent bad data. probably not a problem, just a buggy client.
  console.log(err.message);
});

server.on('listening', function () {
  // fired when all requested servers are listening
  console.log('listening on http port:' + server.http.address().port);
  console.log('listening on udp port:' + server.udp.address().port);
});

server.on('start', function (addr) {
  console.log('got start message from ' + addr);
});

server.listen(0, function() {

});


text:
<!DOCTYPE html>

	<title>onShow / onHide</title>
	<meta http-equiv="Content-Type" content="text/html; charset=UTF-8"/>
	<meta http-equiv="X-UA-Compatible" content="IE=edge"/>
	<link rel="stylesheet" type="text/css" href="../../../codebase/fonts/font_roboto/roboto.css"/>
	<link rel="stylesheet" type="text/css" href="../../../codebase/dhtmlx.css"/>

<input type="text" id="cal_1">


text:
package com.google.cloud.tools.eclipse.appengine.newproject.maven;

import com.google.cloud.tools.eclipse.appengine.newproject.Messages;
import org.eclipse.core.runtime.IStatus;
import org.eclipse.core.runtime.Status;
import org.eclipse.jface.dialogs.DialogPage;
import org.eclipse.jface.dialogs.IMessageProvider;
import org.eclipse.jface.layout.GridLayoutFactory;
import org.eclipse.swt.SWT;
import org.eclipse.swt.events.ModifyListener;
import org.eclipse.swt.events.SelectionAdapter;
import org.eclipse.swt.events.SelectionEvent;
import org.eclipse.swt.widgets.Button;
import org.eclipse.swt.widgets.Composite;
import org.eclipse.swt.widgets.Listener;

public class MavenCoordinatesWizardUi extends Composite {

  private final Button asMavenProjectButton;

  private final MavenCoordinatesUi mavenCoordinatesUi;

  public MavenCoordinatesWizardUi(Composite container, int style) {
    super(container, style);

    asMavenProjectButton = new Button(this, SWT.CHECK);
    asMavenProjectButton.setText(Messages.getString("CREATE_AS_MAVEN_PROJECT")); //$NON-NLS-1$
    asMavenProjectButton.addSelectionListener(new SelectionAdapter() {
      @Override
      public void widgetSelected(SelectionEvent event) {
        updateEnablement();
      }
    });

    mavenCoordinatesUi = new MavenCoordinatesUi(this, SWT.NONE);
    mavenCoordinatesUi.setText(Messages.getString("MAVEN_PROJECT_COORDINATES")); //$NON-NLS-1$

    updateEnablement();

    GridLayoutFactory.swtDefaults().generateLayout(this);
  }

  public boolean uiEnabled() {
    return asMavenProjectButton.getSelection();
  }

  public String getGroupId() {
    return mavenCoordinatesUi.getGroupId();
  }

  public String getArtifactId() {
    return mavenCoordinatesUi.getArtifactId();
  }

  public String getVersion() {
    return mavenCoordinatesUi.getVersion();
  }

  public void addChangeListener(Listener listener) {
    mavenCoordinatesUi.addChangeListener(listener);
    if (asMavenProjectButton != null) {
      asMavenProjectButton.addListener(SWT.Selection, listener);
    }
  }

  public void addGroupIdModifyListener(ModifyListener listener) {
    mavenCoordinatesUi.addGroupIdModifyListener(listener);
  }

  private void updateEnablement() {
    boolean checked = asMavenProjectButton.getSelection();
    mavenCoordinatesUi.setEnabled(checked);
  }

  /**
   * @return {@link IStatus#OK} if there was no validation problem or the UI is disabled; otherwise
   *     a status describing a validation problem (with a non-OK status)
   */
  public IStatus validateMavenSettings() {
    if (!uiEnabled()) {
      return Status.OK_STATUS;
    } else {
      return mavenCoordinatesUi.validateMavenSettings();
    }
  }

  /**
   * Convenience method to set a validation message on {@link DialogPage} from the result of calling
   * {@link #validateMavenSettings()}.
   *
   * @return {@code true} if no validation message was set; {@code false} otherwise
   *
   * @see #validateMavenSettings()
   */
  public boolean setValidationMessage(DialogPage page) {
    IStatus status = validateMavenSettings();
    if (status.isOK()) {
      return true;
    }

    if (IStatus.ERROR == status.getSeverity()) {
      page.setErrorMessage(status.getMessage());
    } else if (IStatus.WARNING == status.getSeverity()) {
      page.setMessage(status.getMessage(), IMessageProvider.WARNING);
    } else if (IStatus.INFO == status.getSeverity()) {
      page.setMessage(status.getMessage(), IMessageProvider.INFORMATION);
    }
    return false;
  }
}