# Transforms.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.nio.ByteBuffer;
    import java.nio.charset.Charset;
    import java.text.MessageFormat;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.Lock;
    import java.util.logging.Logger;
    import org.qi4j.functional.Function;
    import org.qi4j.functional.Specification;

    /**
     * Utility class for I/O transforms
     */
    public class Transforms
    {
        /**
         * Filter items in a transfer by applying the given Specification to each item.
         *
         * @param specification            The Specification defining the items to not filter away.
         * @param output                   The Output instance to receive to result.
         * @param <T>                      The item type
         * @param <Receiver2ThrowableType> Exception type that might be thrown by the Receiver.
         *
         * @return And Output encapsulation the filter operation.
         */
        public static <T, Receiver2ThrowableType extends Throwable> Output<T, Receiver2ThrowableType> filter( final Specification<? super T> specification,
                                                                                                              final Output<T, Receiver2ThrowableType> output
        )
        {
            return new Output<T, Receiver2ThrowableType>()
            {
                @Override
                public <SenderThrowableType extends Throwable> void receiveFrom( final Sender<? extends T, SenderThrowableType> sender )
                    throws Receiver2ThrowableType, SenderThrowableType
                {
                    output.receiveFrom( new Sender<T, SenderThrowableType>()
                    {
                        @Override
                        public <ReceiverThrowableType extends Throwable> void sendTo( final Receiver<? super T, ReceiverThrowableType> receiver )
                            throws ReceiverThrowableType, SenderThrowableType
                        {
                            sender.sendTo( new Receiver<T, ReceiverThrowableType>()
                            {
                                @Override
                                public void receive( T item )
                                    throws ReceiverThrowableType
                                {
                                    if( specification.satisfiedBy( item ) )
                                    {
                                        receiver.receive( item );
                                    }
                                }
                            } );
                        }
                    } );
                }
            };
        }

        /**
         * Map items in a transfer from one type to another by applying the given function.
         *
         * @param function                 The transformation function to apply to the streaming items.
         * @param output                   The output to receive the transformed items.
         * @param <From>                   The type of the incoming items.
         * @param <To>                     The type of the transformed items.
         * @param <Receiver2ThrowableType> The exception type that the Receiver might throw.
         *
         * @return An Output instance that encapsulates the map transformation.
         */
        public static <From, To, Receiver2ThrowableType extends Throwable> Output<From, Receiver2ThrowableType> map( final Function<? super From, ? extends To> function,
                                                                                                                     final Output<To, Receiver2ThrowableType> output
        )
        {
            return new Output<From, Receiver2ThrowableType>()
            {
                @Override
                public <SenderThrowableType extends Throwable> void receiveFrom( final Sender<? extends From, SenderThrowableType> sender )
                    throws Receiver2ThrowableType, SenderThrowableType
                {
                    output.receiveFrom( new Sender<To, SenderThrowableType>()
                    {
                        @Override
                        public <ReceiverThrowableType extends Throwable> void sendTo( final Receiver<? super To, ReceiverThrowableType> receiver )
                            throws ReceiverThrowableType, SenderThrowableType
                        {
                            sender.sendTo( new Receiver<From, ReceiverThrowableType>()
                            {
                                @Override
                                public void receive( From item )
                                    throws ReceiverThrowableType
                                {
                                    receiver.receive( function.map( item ) );
                                }
                            } );
                        }
                    } );
                }
            };
        }

        /**
         * Apply the given function to items in the transfer that match the given specification. Other items will pass
         * through directly.
         *
         * @param specification            The Specification defining which items should be transformed.
         * @param function                 The transformation function.
         * @param output                   The Output that will receive the resulting items.
         * @param <T>                      The item type. Items can not be transformed to a new type.
         * @param <Receiver2ThrowableType> The exception that the Receiver might throw.
         *
         * @return An Output instance that encapsulates the operation.
         */
        public static <T, Receiver2ThrowableType extends Throwable> Output<T, Receiver2ThrowableType> filteredMap( final Specification<? super T> specification,
                                                                                                                   final Function<? super T, ? extends T> function,
                                                                                                                   final Output<T, Receiver2ThrowableType> output
        )
        {
            return new Output<T, Receiver2ThrowableType>()
            {
                @Override
                public <SenderThrowableType extends Throwable> void receiveFrom( final Sender<? extends T, SenderThrowableType> sender )
                    throws Receiver2ThrowableType, SenderThrowableType
                {
                    output.receiveFrom( new Sender<T, SenderThrowableType>()
                    {
                        @Override
                        public <ReceiverThrowableType extends Throwable> void sendTo( final Receiver<? super T, ReceiverThrowableType> receiver )
                            throws ReceiverThrowableType, SenderThrowableType
                        {
                            sender.sendTo( new Receiver<T, ReceiverThrowableType>()
                            {
                                @Override
                                public void receive( T item )
                                    throws ReceiverThrowableType
                                {
                                    if( specification.satisfiedBy( item ) )
                                    {
                                        receiver.receive( function.map( item ) );
                                    }
                                    else
                                    {
                                        receiver.receive( item );
                                    }
                                }
                            } );
                        }
                    } );
                }
            };
        }

        /**
         * Wrapper for Outputs that uses a lock whenever a transfer is instantiated. Typically a read-lock would be used on
         * the sending side and a write-lock would be used on the receiving side. Inputs can use this as well to create a
         * wrapper on the send side when transferTo is invoked.
         *
         * @param lock                    the lock to be used for transfers
         * @param output                  output to be wrapped
         * @param <T>                     The Item type
         * @param <Receiver2ThrowableType> The Exception type that the Receiver might throw.
         *
         * @return Output wrapper that uses the given lock during transfers.
         */
        public static <T, Receiver2ThrowableType extends Throwable> Output<T, Receiver2ThrowableType> lock( final Lock lock,
                                                                                                          final Output<T, Receiver2ThrowableType> output
        )
        {
            return new Output<T, Receiver2ThrowableType>()
            {
                @Override
                public <SenderThrowableType extends Throwable> void receiveFrom( Sender<? extends T, SenderThrowableType> sender )
                    throws Receiver2ThrowableType, SenderThrowableType
                {
                    /**
                     * Fix for this bug:
                     * http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6822370
                     */
                    while( true )
                    {
                        try
                        {
                            //noinspection StatementWithEmptyBody
                            while( !lock.tryLock( 1000, TimeUnit.MILLISECONDS ) )
                            {
                                // On timeout, try again
                            }
                            break; // Finally got a lock
                        }
                        catch( InterruptedException e )
                        {
                            // Try again
                        }
                    }

                    try
                    {
                        output.receiveFrom( sender );
                    }
                    finally
                    {
                        lock.unlock();
                    }
                }
            };
        }

        /**
         * Wrapper for Outputs that uses a lock whenever a transfer is instantiated. Typically a read-lock would be used on the sending side and a write-lock
         * would be used on the receiving side.
         *
         * @param lock                  the lock to be used for transfers
         * @param input                 input to be wrapped
         * @param <T>                   The item type.
         * @param <SenderThrowableType> The Exception type that the Sender might throw.
         *
         * @return Input wrapper that uses the given lock during transfers.
         */
        public static <T, SenderThrowableType extends Throwable> Input<T, SenderThrowableType> lock( final Lock lock,
                                                                                                     final Input<T, SenderThrowableType> input
        )
        {
            return new Input<T, SenderThrowableType>()
            {
                @Override
                public <ReceiverThrowableType extends Throwable> void transferTo( Output<? super T, ReceiverThrowableType> output )
                    throws SenderThrowableType, ReceiverThrowableType
                {
                    /**
                     * Fix for this bug:
                     * http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=6822370
                     */
                    while( true )
                    {
                        try
                        {
                            //noinspection StatementWithEmptyBody
                            while( !( lock.tryLock() || lock.tryLock( 1000, TimeUnit.MILLISECONDS ) ) )
                            {
                                // On timeout, try again
                            }
                            break; // Finally got a lock
                        }
                        catch( InterruptedException e )
                        {
                            // Try again
                        }
                    }

                    try
                    {
                        input.transferTo( output );
                    }
                    finally
                    {
                        lock.unlock();
                    }
                }
            };
        }

        /**
         * Count the number of items in the transfer.
         *
         * @param <T>
         */
        // START SNIPPET: counter
        public static class Counter<T>
            implements Function<T, T>
        {
            private volatile long count = 0;

            public long count()
            {
                return count;
            }

            @Override
            public T map( T t )
            {
                count++;
                return t;
            }
        }
        // END SNIPPET: counter

        /**
         * Convert strings to bytes using the given CharSet
         */
        @SuppressWarnings( "UnusedDeclaration" )
        public static class String2Bytes
            implements Function<String, byte[]>
        {
            private Charset charSet;

            public String2Bytes( Charset charSet )
            {
                this.charSet = charSet;
            }

            @Override
            public byte[] map( String s )
            {
                return s.getBytes( charSet );
            }
        }

        /**
         * Convert ByteBuffers to Strings using the given CharSet
         */
        public static class ByteBuffer2String
            implements Function<ByteBuffer, String>
        {
            private Charset charSet;

            public ByteBuffer2String( Charset charSet )
            {
                this.charSet = charSet;
            }

            @Override
            public String map( ByteBuffer buffer )
            {
                return new String( buffer.array(), charSet );
            }
        }

        /**
         * Convert objects to Strings using .toString()
         */
        @SuppressWarnings( "UnusedDeclaration" )
        public static class ObjectToString
            implements Function<Object, String>
        {
            @Override
            public String map( Object o )
            {
                return o.toString();
            }
        }

        /**
         * Log the toString() representation of transferred items to the given log. The string is first formatted using MessageFormat
         * with the given format.
         *
         * @param <T>
         */
        public static class Log<T>
            implements Function<T, T>
        {
            private Logger logger;
            private MessageFormat format;

            public Log( Logger logger, String format )
            {
                this.logger = logger;
                this.format = new MessageFormat( format );
            }

            @Override
            public T map( T item )
            {
                logger.info( format.format( new String[]{ item.toString() } ) );
                return item;
            }
        }

        /**
         * Track progress of transfer by emitting a log message in given intervals.
         *
         * If logger or format is null, then you need to override the logProgress to do something
         *
         * @param <T> type of items to be transferred
         */
        // START SNIPPET: progress
        public static class ProgressLog<T>
            implements Function<T, T>
        {
            private Counter<T> counter;
            private Log<String> log;
            private final long interval;

            public ProgressLog( Logger logger, String format, long interval )
            {
                this.interval = interval;
                if( logger != null && format != null )
                {
                    log = new Log<>( logger, format );
                }
                counter = new Counter<>();
            }

            public ProgressLog( long interval )
            {
                this.interval = interval;
                counter = new Counter<>();
            }

            @Override
            public T map( T t )
            {
                counter.map( t );
                if( counter.count % interval == 0 )
                {
                    logProgress();
                }
                return t;
            }

            // Override this to do something other than logging the progress
            protected void logProgress()
            {
                if( log != null )
                {
                    log.map( counter.count + "" );
                }
            }
        }
        // END SNIPPET: progress
    }