core/io/src/main/java/org/qi4j/io/Inputs.java

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

import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.Reader;
import java.net.URL;
import java.net.URLConnection;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Scanner;
import java.util.zip.GZIPInputStream;
import org.qi4j.functional.Visitor;

/**
 * Common inputs
 */
public class Inputs
{
    // START SNIPPET: method

    /**
     * Read lines from a String.
     *
     * @param source lines
     *
     * @return Input that provides lines from the string as strings
     */
    public static Input<String, RuntimeException> text( final String source )
    // END SNIPPET: method
    {
        return new Input<String, RuntimeException>()
        {
            @Override
            public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super String, ReceiverThrowableType> output )
                throws RuntimeException, ReceiverThrowableType
            {

                output.receiveFrom( new Sender<String, RuntimeException>()
                {
                    @Override
                    public <Receiver2ThrowableType extends Throwable> void sendTo( Receiver<? super String, Receiver2ThrowableType> receiver )
                        throws Receiver2ThrowableType, RuntimeException
                    {
                        Scanner scanner = new Scanner( source );
                        while( scanner.hasNextLine() )
                        {
                            receiver.receive( scanner.nextLine() );
                        }
                    }
                } );
            }
        };
    }

    // START SNIPPET: method

    /**
     * Read lines from a Reader.
     *
     * @param source lines
     *
     * @return Input that provides lines from the string as strings
     */
    public static Input<String, RuntimeException> text( final Reader source )
    // END SNIPPET: method
    {
        return new Input<String, RuntimeException>()
        {
            @Override
            public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super String, ReceiverThrowableType> output )
                throws RuntimeException, ReceiverThrowableType
            {

                output.receiveFrom( new Sender<String, RuntimeException>()
                {
                    @Override
                    public <Receiver2ThrowableType extends Throwable> void sendTo( Receiver<? super String, Receiver2ThrowableType> receiver )
                        throws Receiver2ThrowableType, RuntimeException
                    {
                        Scanner scanner = new Scanner( source );
                        while( scanner.hasNextLine() )
                        {
                            receiver.receive( scanner.nextLine() );
                        }
                    }
                } );
            }
        };
    }

    // START SNIPPET: method

    /**
     * Read lines from a UTF-8 encoded textfile.
     *
     * If the filename ends with .gz, then the data is automatically unzipped when read.
     *
     * @param source textfile with lines separated by \n character
     *
     * @return Input that provides lines from the textfiles as strings
     */
    public static Input<String, IOException> text( final File source )
    // END SNIPPET: method
    {
        return text( source, "UTF-8" );
    }

    // START SNIPPET: method

    /**
     * Read lines from a textfile with the given encoding.
     *
     * If the filename ends with .gz, then the data is automatically unzipped when read.
     *
     * @param source   textfile with lines separated by \n character
     * @param encoding encoding of file, e.g. "UTF-8"
     *
     * @return Input that provides lines from the textfiles as strings
     */
    public static Input<String, IOException> text( final File source, final String encoding )
    // END SNIPPET: method
    {
        return new Input<String, IOException>()
        {
            @Override
            public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super String, ReceiverThrowableType> output )
                throws IOException, ReceiverThrowableType
            {
                InputStream stream = new FileInputStream( source );

                // If file is gzipped, unzip it automatically
                if( source.getName().endsWith( ".gz" ) )
                {
                    stream = new GZIPInputStream( stream );
                }

                try (BufferedReader reader = new BufferedReader( new InputStreamReader( stream, encoding ) ))
                {
                    output.receiveFrom( new Sender<String, IOException>()
                    {
                        @Override
                        public <Receiver2ThrowableType extends Throwable> void sendTo( Receiver<? super String, Receiver2ThrowableType> receiver )
                            throws Receiver2ThrowableType, IOException
                        {
                            String line;
                            while( ( line = reader.readLine() ) != null )
                            {
                                receiver.receive( line );
                            }
                        }
                    } );
                }
            }
        };
    }

    // START SNIPPET: method

    /**
     * Read lines from a textfile at a given URL.
     *
     * If the content support gzip encoding, then the data is automatically unzipped when read.
     *
     * The charset in the content-type of the URL will be used for parsing. Default is UTF-8.
     *
     * @param source textfile with lines separated by \n character
     *
     * @return Input that provides lines from the textfiles as strings
     */
    public static Input<String, IOException> text( final URL source )
    // END SNIPPET: method
    {
        return new Input<String, IOException>()
        {
            @Override
            public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super String, ReceiverThrowableType> output )
                throws IOException, ReceiverThrowableType
            {
                URLConnection urlConnection = source.openConnection();
                urlConnection.setRequestProperty( "Accept-Encoding", "gzip" );
                InputStream stream = urlConnection.getInputStream();

                // If file is gzipped, unzip it automatically
                if( "gzip".equals( urlConnection.getContentEncoding() ) )
                {
                    stream = new GZIPInputStream( stream );
                }

                // Figure out charset given content-type
                String contentType = urlConnection.getContentType();
                String charSet = "UTF-8";
                if( contentType.contains( "charset=" ) )
                {
                    charSet = contentType.substring( contentType.indexOf( "charset=" ) + "charset=".length() );
                }

                try (BufferedReader reader = new BufferedReader( new InputStreamReader( stream, charSet ) ))
                {
                    output.receiveFrom( new Sender<String, IOException>()
                    {
                        @Override
                        public <Receiver2ThrowableType extends Throwable> void sendTo( Receiver<? super String, Receiver2ThrowableType> receiver )
                            throws Receiver2ThrowableType, IOException
                        {
                            String line;
                            while( ( line = reader.readLine() ) != null )
                            {
                                receiver.receive( line );
                            }
                        }
                    } );
                }
            }
        };
    }

    // START SNIPPET: method

    /**
     * Read a file using ByteBuffer of a given size. Useful for transferring raw data.
     *
     * @param source The file to be read.
     * @param bufferSize The size of the byte array.
     *
     * @return An Input instance to be applied to streaming operations.
     */
    public static Input<ByteBuffer, IOException> byteBuffer( final File source, final int bufferSize )
    // END SNIPPET: method
    {
        return new Input<ByteBuffer, IOException>()
        {
            @Override
            public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super ByteBuffer, ReceiverThrowableType> output )
                throws IOException, ReceiverThrowableType
            {
                final FileInputStream stream = new FileInputStream( source );
                final FileChannel fci = stream.getChannel();

                final ByteBuffer buffer = ByteBuffer.allocate( bufferSize );

                try
                {
                    output.receiveFrom( new Sender<ByteBuffer, IOException>()
                    {
                        @Override
                        public <Receiver2ThrowableType extends Throwable> void sendTo( Receiver<? super ByteBuffer, Receiver2ThrowableType> receiver )
                            throws Receiver2ThrowableType, IOException
                        {
                            while( fci.read( buffer ) != -1 )
                            {
                                buffer.flip();
                                receiver.receive( buffer );
                                buffer.clear();
                            }
                        }
                    } );
                }
                finally
                {
                    stream.close();
                }
            }
        };
    }

    // START SNIPPET: method

    /**
     * Read an inputstream using ByteBuffer of a given size.
     *
     * @param source The InputStream to be read.
     * @param bufferSize The size of the byte array.
     *
     * @return An Input instance to be applied to streaming operations.
     */
    public static Input<ByteBuffer, IOException> byteBuffer( final InputStream source, final int bufferSize )
    // END SNIPPET: method
    {
        return new Input<ByteBuffer, IOException>()
        {
            @Override
            public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super ByteBuffer, ReceiverThrowableType> output )
                throws IOException, ReceiverThrowableType
            {
                try
                {
                    output.receiveFrom( new Sender<ByteBuffer, IOException>()
                    {
                        @Override
                        public <Receiver2ThrowableType extends Throwable> void sendTo( Receiver<? super ByteBuffer, Receiver2ThrowableType> receiver )
                            throws Receiver2ThrowableType, IOException
                        {
                            byte[] buffer = new byte[ bufferSize ];

                            int len;
                            while( ( len = source.read( buffer ) ) != -1 )
                            {
                                ByteBuffer byteBuffer = ByteBuffer.wrap( buffer, 0, len );
                                receiver.receive( byteBuffer );
                            }
                        }
                    } );
                }
                finally
                {
                    source.close();
                }
            }
        };
    }

    // START SNIPPET: method

    /**
     * Combine many Input into one single Input. When a transfer is initiated from it all items from all inputs will be transferred
     * to the given Output.
     *
     * @param inputs An Iterable of Input instances to be combined.
     * @param <T> The item type of the Input
     * @param <SenderThrowableType> The Throwable that might be thrown by the Inputs.
     *
     * @return A combined Input, allowing for easy aggregation of many Input sources.
     */
    public static <T, SenderThrowableType extends Throwable> Input<T, SenderThrowableType> combine( final Iterable<Input<T, SenderThrowableType>> inputs )
    // END SNIPPET: method
    {
        return new Input<T, SenderThrowableType>()
        {
            @Override
            public <Receiver2ThrowableType extends Throwable> void transferTo( Output<? super T, Receiver2ThrowableType> output )
                throws SenderThrowableType, Receiver2ThrowableType
            {
                output.receiveFrom( new Sender<T, SenderThrowableType>()
                {
                    @Override
                    public <ReceiverThrowableType extends Throwable> void sendTo( final Receiver<? super T, ReceiverThrowableType> receiver )
                        throws ReceiverThrowableType, SenderThrowableType
                    {
                        for( Input<T, SenderThrowableType> input : inputs )
                        {
                            input.transferTo( new Output<T, ReceiverThrowableType>()
                            {
                                @Override
                                public <Sender2ThrowableType extends Throwable> void receiveFrom( Sender<? extends T, Sender2ThrowableType> sender )
                                    throws ReceiverThrowableType, Sender2ThrowableType
                                {
                                    sender.sendTo( new Receiver<T, ReceiverThrowableType>()
                                    {
                                        @Override
                                        public void receive( T item )
                                            throws ReceiverThrowableType
                                        {
                                            receiver.receive( item );
                                        }
                                    } );
                                }
                            } );
                        }
                    }
                } );
            }
        };
    }

    // START SNIPPET: method

    /**
     * Create an Input that takes its items from the given Iterable.
     *
     * @param iterable The Iterable to be used as an Input.
     * @param <T> The item type of the Input
     *
     * @return An Input instance that is backed by the Iterable.
     */
    public static <T> Input<T, RuntimeException> iterable( final Iterable<T> iterable )
    // END SNIPPET: method
    {
        return new Input<T, RuntimeException>()
        {
            @Override
            public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super T, ReceiverThrowableType> output )
                throws RuntimeException, ReceiverThrowableType
            {
                output.receiveFrom( new Sender<T, RuntimeException>()
                {
                    @Override
                    public <Receiver2ThrowableType extends Throwable> void sendTo( Receiver<? super T, Receiver2ThrowableType> receiver )
                        throws Receiver2ThrowableType, RuntimeException
                    {
                        for( T item : iterable )
                        {
                            receiver.receive( item );
                        }
                    }
                } );
            }
        };
    }

    // START SNIPPET: method

    /**
     * Create an Input that allows a Visitor to write to an OutputStream. The stream is a BufferedOutputStream, so when enough
     * data has been gathered it will send this in chunks of the given size to the Output it is transferred to. The Visitor does not have to call
     * close() on the OutputStream, but should ensure that any wrapper streams or writers are flushed so that all data is sent.
     *
     * @param outputVisitor The OutputStream Visitor that will be backing the Input ByteBuffer.
     * @param bufferSize The buffering size.
     *
     * @return An Input instance of ByteBuffer, that is backed by an Visitor to an OutputStream.
     */
    public static Input<ByteBuffer, IOException> output( final Visitor<OutputStream, IOException> outputVisitor,
                                                         final int bufferSize
    )
    // END SNIPPET: method
    {
        return new Input<ByteBuffer, IOException>()
        {
            @Override
            public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super ByteBuffer, ReceiverThrowableType> output )
                throws IOException, ReceiverThrowableType
            {
                output.receiveFrom( new Sender<ByteBuffer, IOException>()
                {
                    @Override
                    @SuppressWarnings( "unchecked" )
                    public <Receiver2ThrowableType extends Throwable> void sendTo( final Receiver<? super ByteBuffer, Receiver2ThrowableType> receiver )
                        throws Receiver2ThrowableType, IOException
                    {
                        try (OutputStream out = new BufferedOutputStream( new OutputStream()
                        {
                            @Override
                            public void write( int b )
                                throws IOException
                            {
                                // Ignore
                            }

                            @SuppressWarnings( "NullableProblems" )
                            @Override
                            public void write( byte[] b, int off, int len )
                                throws IOException
                            {
                                try
                                {
                                    ByteBuffer byteBuffer = ByteBuffer.wrap( b, 0, len );
                                    receiver.receive( byteBuffer );
                                }
                                catch( Throwable ex )
                                {
                                    throw new IOException( ex );
                                }
                            }
                        }, bufferSize ))
                        {
                            outputVisitor.visit( out );
                        }
                        catch( IOException ex )
                        {
                            throw (Receiver2ThrowableType) ex.getCause();
                        }
                    }
                } );
            }
        };
    }

    private Inputs()
    {
    }
}