<commit_before>/*
 * Copyright (c) 2002-2015 "Neo Technology,"
 * Network Engine for Objects in Lund AB [http://neotechnology.com]
 *
 * This file is part of Neo4j.
 *
 * Neo4j is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.neo4j.test.subprocess;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintStream;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Proxy;
import java.lang.reflect.Type;
import java.nio.ByteBuffer;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.rmi.ServerError;
import java.rmi.server.RemoteObject;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.neo4j.function.Predicate;
import org.neo4j.test.ProcessStreamHandler;

@SuppressWarnings( "serial" )
public abstract class SubProcess<T, P> implements Serializable
{
    private interface NoInterface
    {
        // Used when no interface is declared
    }

    // by default source and destination for subprocess standard I/O to be the same as those of the current process
    private static final boolean INHERIT_IO_DEFAULT_VALUE = true;

    private final Class<T> t;
    private transient boolean inheritIo = INHERIT_IO_DEFAULT_VALUE;
    private final transient Predicate<String> classPathFilter;

    @SuppressWarnings( { "unchecked", "rawtypes" } )
    public SubProcess( Predicate<String> classPathFilter, boolean inheritIo )
    {
        this.inheritIo = inheritIo;
        if ( getClass().getSuperclass() != SubProcess.class )
        {
            throw new ClassCastException( SubProcess.class.getName() + " may only be extended one level " );
        }
        Class<?> me = getClass();
        while ( me.getSuperclass() != SubProcess.class )
        {
            me = me.getSuperclass();
        }
        Type type = ( (ParameterizedType) me.getGenericSuperclass() ).getActualTypeArguments()[0];
        @SuppressWarnings( { "hiding" } ) Class<T> t;
        if ( type instanceof Class<?> )
        {
            t = (Class<T>) type;
        }
        else if ( type instanceof ParameterizedType )
        {
            t = (Class<T>) ( (ParameterizedType) type ).getRawType();
        }
        else
        {
            throw new ClassCastException( "Illegal type parameter " + type );
        }
        if ( t == Object.class )
        {
            t = (Class) NoInterface.class;
        }
        if ( !t.isInterface() )
        {
            throw new ClassCastException( t + " is not an interface" );
        }
        if ( t.isAssignableFrom( getClass() ) || t == NoInterface.class )
        {
            this.t = t;
        }
        else
        {
            throw new ClassCastException( getClass().getName() + " must implement declared interface " + t );
        }
        this.classPathFilter = classPathFilter;
    }

    public SubProcess( Predicate<String> classPathFilter )
    {
        this( classPathFilter, INHERIT_IO_DEFAULT_VALUE );
    }

    public SubProcess( boolean inheritIo )
    {
        this(null, inheritIo);
    }

    public SubProcess()
    {
        this( null, INHERIT_IO_DEFAULT_VALUE );
    }

    public T start( P parameter, BreakPoint... breakpoints )
    {
        DebuggerConnector debugger = null;
        if ( breakpoints != null && breakpoints.length != 0 )
        {
            debugger = new DebuggerConnector( breakpoints );
        }
        DispatcherTrapImpl callback;
        try
        {
            callback = new DispatcherTrapImpl( this, parameter );
        }
        catch ( RemoteException e )
        {
            throw new RuntimeException( "Failed to create local RMI endpoint.", e );
        }
        Process process;
        String pid;
        DebugDispatch debugDispatch = null;
        Dispatcher dispatcher;
        try
        {
            synchronized ( debugger != null ? DebuggerConnector.class : new Object() )
            {
                if ( debugger != null )
                {
                    process = start(inheritIo, "java", "-ea", "-Xmx1G", debugger.listen(),
                            "-Djava.awt.headless=true", "-cp",
                            classPath( System.getProperty( "java.class.path" ) ), SubProcess.class.getName(),
                            serialize( callback ) );
                }
                else
                {
                    process = start(inheritIo, "java", "-ea", "-Xmx1G", "-Djava.awt.headless=true", "-cp",
                            classPath( System.getProperty( "java.class.path" ) ),
                            SubProcess.class.getName(), serialize( callback ) );
                }
                pid = getPid( process );
                // if IO was not inherited by current process we need to pipe error and input stream to corresponding
                // target streams
                if (!inheritIo)
                {
                    pipe( "[" + toString() + ":" + pid + "] ", process.getErrorStream(), errorStreamTarget() );
                    pipe( "[" + toString() + ":" + pid + "] ", process.getInputStream(), inputStreamTarget() );
                }
                if ( debugger != null )
                {
                    debugDispatch = debugger.connect( toString() + ":" + pid );
                }
            }
            dispatcher = callback.get( process );
        }
        finally
        {
            try
            {
                UnicastRemoteObject.unexportObject( callback, true );
            }
            catch ( RemoteException e )
            {
                e.printStackTrace();
            }
        }
        if ( dispatcher == null )
        {
            throw new IllegalStateException( "failed to start sub process" );
        }
        Handler handler = new Handler( t, dispatcher, process, "<" + toString() + ":" + pid + ">", debugDispatch );
        if ( debugDispatch != null )
        {
            debugDispatch.handler = handler;
        }
        return t.cast( Proxy.newProxyInstance( t.getClassLoader(), new Class[] { t }, live( handler ) ) );
    }

    protected PrintStream errorStreamTarget()
    {
        return System.err;
    }

    protected PrintStream inputStreamTarget()
    {
        return System.out;
    }

    private String classPath( String parentClasspath )
    {
        if ( classPathFilter == null )
        {
            return parentClasspath;
        }
        StringBuilder result = new StringBuilder();
        for ( String part : parentClasspath.split( File.pathSeparator ) )
        {
            if ( classPathFilter.test( part ) )
            {
                result.append( result.length() > 0 ? File.pathSeparator : "" ).append( part );
            }
        }
        return result.toString();
    }

    private static Process start(boolean inheritIo, String... args )
    {
        ProcessBuilder builder = new ProcessBuilder( args );
        if ( inheritIo )
        {
            builder.inheritIO();
        }
        try
        {
            return builder.start();
        }
        catch ( IOException e )
        {
            throw new RuntimeException( "Failed to start sub process", e );
        }
    }

    @SuppressWarnings( "restriction" )<commit_msg>Subprocess by default will inherit only output destinations of main process.<commit_after>/*
 * Copyright (c) 2002-2015 "Neo Technology,"
 * Network Engine for Objects in Lund AB [http://neotechnology.com]
 *
 * This file is part of Neo4j.
 *
 * Neo4j is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */
package org.neo4j.test.subprocess;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintStream;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Proxy;
import java.lang.reflect.Type;
import java.nio.ByteBuffer;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.rmi.ServerError;
import java.rmi.server.RemoteObject;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.neo4j.function.Predicate;
import org.neo4j.test.ProcessStreamHandler;

@SuppressWarnings( "serial" )
public abstract class SubProcess<T, P> implements Serializable
{
    private interface NoInterface
    {
        // Used when no interface is declared
    }

    // by default will inherit output destinations for subprocess from current process
    private static final boolean INHERIT_OUTPUT_DEFAULT_VALUE = true;

    private final Class<T> t;
    private transient boolean inheritOutput = INHERIT_OUTPUT_DEFAULT_VALUE;
    private final transient Predicate<String> classPathFilter;

    @SuppressWarnings( { "unchecked", "rawtypes" } )
    public SubProcess( Predicate<String> classPathFilter, boolean inheritOutput )
    {
        this.inheritOutput = inheritOutput;
        if ( getClass().getSuperclass() != SubProcess.class )
        {
            throw new ClassCastException( SubProcess.class.getName() + " may only be extended one level " );
        }
        Class<?> me = getClass();
        while ( me.getSuperclass() != SubProcess.class )
        {
            me = me.getSuperclass();
        }
        Type type = ( (ParameterizedType) me.getGenericSuperclass() ).getActualTypeArguments()[0];
        @SuppressWarnings( { "hiding" } ) Class<T> t;
        if ( type instanceof Class<?> )
        {
            t = (Class<T>) type;
        }
        else if ( type instanceof ParameterizedType )
        {
            t = (Class<T>) ( (ParameterizedType) type ).getRawType();
        }
        else
        {
            throw new ClassCastException( "Illegal type parameter " + type );
        }
        if ( t == Object.class )
        {
            t = (Class) NoInterface.class;
        }
        if ( !t.isInterface() )
        {
            throw new ClassCastException( t + " is not an interface" );
        }
        if ( t.isAssignableFrom( getClass() ) || t == NoInterface.class )
        {
            this.t = t;
        }
        else
        {
            throw new ClassCastException( getClass().getName() + " must implement declared interface " + t );
        }
        this.classPathFilter = classPathFilter;
    }

    public SubProcess( Predicate<String> classPathFilter )
    {
        this( classPathFilter, INHERIT_OUTPUT_DEFAULT_VALUE );
    }

    public SubProcess( boolean inheritOutput )
    {
        this(null, inheritOutput );
    }

    public SubProcess()
    {
        this( null, INHERIT_OUTPUT_DEFAULT_VALUE );
    }

    public T start( P parameter, BreakPoint... breakpoints )
    {
        DebuggerConnector debugger = null;
        if ( breakpoints != null && breakpoints.length != 0 )
        {
            debugger = new DebuggerConnector( breakpoints );
        }
        DispatcherTrapImpl callback;
        try
        {
            callback = new DispatcherTrapImpl( this, parameter );
        }
        catch ( RemoteException e )
        {
            throw new RuntimeException( "Failed to create local RMI endpoint.", e );
        }
        Process process;
        String pid;
        DebugDispatch debugDispatch = null;
        Dispatcher dispatcher;
        try
        {
            synchronized ( debugger != null ? DebuggerConnector.class : new Object() )
            {
                if ( debugger != null )
                {
                    process = start( inheritOutput, "java", "-ea", "-Xmx1G", debugger.listen(),
                            "-Djava.awt.headless=true", "-cp",
                            classPath( System.getProperty( "java.class.path" ) ), SubProcess.class.getName(),
                            serialize( callback ) );
                }
                else
                {
                    process = start( inheritOutput, "java", "-ea", "-Xmx1G", "-Djava.awt.headless=true", "-cp",
                            classPath( System.getProperty( "java.class.path" ) ),
                            SubProcess.class.getName(), serialize( callback ) );
                }
                pid = getPid( process );
                // if IO was not inherited by current process we need to pipe error and input stream to corresponding
                // target streams
                if (!inheritOutput )
                {
                    pipe( "[" + toString() + ":" + pid + "] ", process.getErrorStream(), errorStreamTarget() );
                    pipe( "[" + toString() + ":" + pid + "] ", process.getInputStream(), inputStreamTarget() );
                }
                if ( debugger != null )
                {
                    debugDispatch = debugger.connect( toString() + ":" + pid );
                }
            }
            dispatcher = callback.get( process );
        }
        finally
        {
            try
            {
                UnicastRemoteObject.unexportObject( callback, true );
            }
            catch ( RemoteException e )
            {
                e.printStackTrace();
            }
        }
        if ( dispatcher == null )
        {
            throw new IllegalStateException( "failed to start sub process" );
        }
        Handler handler = new Handler( t, dispatcher, process, "<" + toString() + ":" + pid + ">", debugDispatch );
        if ( debugDispatch != null )
        {
            debugDispatch.handler = handler;
        }
        return t.cast( Proxy.newProxyInstance( t.getClassLoader(), new Class[] { t }, live( handler ) ) );
    }

    protected PrintStream errorStreamTarget()
    {
        return System.err;
    }

    protected PrintStream inputStreamTarget()
    {
        return System.out;
    }

    private String classPath( String parentClasspath )
    {
        if ( classPathFilter == null )
        {
            return parentClasspath;
        }
        StringBuilder result = new StringBuilder();
        for ( String part : parentClasspath.split( File.pathSeparator ) )
        {
            if ( classPathFilter.test( part ) )
            {
                result.append( result.length() > 0 ? File.pathSeparator : "" ).append( part );
            }
        }
        return result.toString();
    }

    private static Process start(boolean inheritOutput, String... args )
    {
        ProcessBuilder builder = new ProcessBuilder( args );
        if ( inheritOutput )
        {
            // We can not simply use builder.inheritIO here because
            // that will also inherit input which will be closed in case of background execution of main process.
            // Closed input stream will cause immediate exit from a subprocess liveloop.
            // And we use background execution in scripts and on CI server.
            builder.redirectError( ProcessBuilder.Redirect.INHERIT )
                   .redirectOutput( ProcessBuilder.Redirect.INHERIT );
        }
        try
        {
            return builder.start();
        }
        catch ( IOException e )
        {
            throw new RuntimeException( "Failed to start sub process", e );
        }
    }

    @SuppressWarnings( "restriction" )
{'fraction_non_alphanumeric': 0.09839908519153802, 'fraction_numerical': 0.002172670097198399, 'mean_word_length': 2.0027467811158797, 'pattern_counts': {'":': 8, '<': 27, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3635072', 'n_tokens_mistral': 4575, 'n_tokens_neox': 4500, 'n_words': 1282}