<commit_before>/*
 * Licensed to Elastic Search and <NAME> under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. Elastic Search licenses this
 * file to you 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 com.asquera.elasticsearch.river.jruby;

import java.util.Map;
import java.util.List;

import java.net.URL;
import java.net.MalformedURLException;

import java.io.File;

import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.action.bulk.BulkRequestBuilder;
import org.elasticsearch.cluster.block.ClusterBlockException;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.support.XContentMapValues;
import org.elasticsearch.indices.IndexAlreadyExistsException;
import org.elasticsearch.river.AbstractRiverComponent;
import org.elasticsearch.river.River;
import org.elasticsearch.river.RiverName;
import org.elasticsearch.river.RiverSettings;
import org.elasticsearch.threadpool.ThreadPool;

import org.jruby.Ruby;
import org.jruby.RubyModule;
import org.jruby.javasupport.JavaUtil;
import org.jruby.RubyClass;
import org.jruby.embed.ScriptingContainer;
import org.jruby.embed.PathType;
import org.jruby.embed.AttributeName;
import org.jruby.javasupport.util.RuntimeHelpers;
import org.jruby.runtime.builtin.IRubyObject;
import org.jruby.runtime.ThreadContext;
import org.jruby.runtime.GlobalVariable;
import org.jruby.anno.JRubyMethod;
import static org.jruby.runtime.Visibility.*;
import static org.jruby.CompatVersion.*;

/**
 * @author <NAME> (<EMAIL>)
 */
public class JRubyRiver extends AbstractRiverComponent implements River {
    private ScriptingContainer container;
    private IRubyObject river;
    
    @Inject public JRubyRiver(RiverName riverName, RiverSettings settings, Client client, ThreadPool threadPool) throws Exception {        
        super(riverName, settings);
        this.container = new ScriptingContainer();

        if (settings.settings().containsKey("jruby")) {
            Map<String, Object> riverSettings = (Map<String, Object>) settings.settings().get("jruby");
            
            //String className       = XContentMapValues.nodeStringValue(riverSettings.get("ruby_class"), "River");
            String scriptName      = XContentMapValues.nodeStringValue(riverSettings.get("script_name"), "lib/river.rb");
            String scriptDirectory = XContentMapValues.nodeStringValue(riverSettings.get("script_directory"), null);
            
            if (null != scriptDirectory) {
                logger.info("changing for directory [{}]", scriptDirectory);
                container.getLoadPaths().add(scriptDirectory);
                container.setAttribute(AttributeName.BASE_DIR, scriptDirectory);
                container.setCurrentDirectory(scriptDirectory);
            }
            
            Ruby runtime = getRuntime();
            runtime.defineVariable(new GlobalVariable(runtime, "$river", runtime.getNil()));
            RubyModule kernel = runtime.getKernel();
            kernel.defineAnnotatedMethods(JRubyRiver.class);
            
            List<String> loadPath = (List<String>)riverSettings.get("load_path");
            
            if (loadPath != null) {
                for (String path : loadPath) {
                    logger.info(path);
                    container.getLoadPaths().add(path);
                } 
            }

            logger.info("running script [{}]", scriptName);
            container.runScriptlet(PathType.RELATIVE, scriptName);
            logger.info("found river class [{}]", river());
            
            RubyClass klass = river();
            IRubyObject clientInstance = JavaUtil.convertJavaToRuby(runtime, client);
            IRubyObject threadPoolInstance = JavaUtil.convertJavaToRuby(runtime, threadPool);
            IRubyObject settingsInstance = JavaUtil.convertJavaToRuby(runtime, settings);
            IRubyObject nameInstance = JavaUtil.convertJavaToRuby(runtime, riverName);
            IRubyObject loggerInstance = JavaUtil.convertJavaToRuby(runtime, logger);
            
            this.river = RuntimeHelpers.invoke(runtime.getCurrentContext(), klass, "new", nameInstance, settingsInstance, clientInstance, threadPoolInstance, loggerInstance);
        } else {
            throw new Exception("No options for jruby river found");
        }
    }
    
    private URL getURL(String target) throws MalformedURLException {
        try {
            // First try assuming a protocol is included
            return new URL(target);
        } catch (MalformedURLException e) {
            // Assume file: protocol
            File f = new File(target);
            String path = target;
            if (f.exists() && f.isDirectory() && !path.endsWith("/")) {
                // URLClassLoader requires that directores end with slashes
                path = path + "/";
            }
            return new URL("file", null, path);
        }
    }

    @Override public void start() {
        RuntimeHelpers.invoke(getRuntime().getCurrentContext(), this.river, "start");
    }

    @Override public void close() {
        RuntimeHelpers.invoke(getRuntime().getCurrentContext(), this.river, "close");
    }
    
    private Ruby getRuntime() {
        return container.getProvider().getRuntime();
    }
    
    private RubyClass river() {
        return (RubyClass)getRuntime().getGlobalVariables().get("$river");
    }
    
    @JRubyMethod(name = "river", required = 1, visibility = PRIVATE)
    public static void river(ThreadContext context, IRubyObject recv, IRubyObject object) {
        context.getRuntime().getGlobalVariables().set("$river", object);
    }
<commit_msg>Use SINGLETHREAD to enable one Container per River<commit_after>/*
 * Licensed to Elastic Search and <NAME> under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. Elastic Search licenses this
 * file to you 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 com.asquera.elasticsearch.river.jruby;

import java.util.Map;
import java.util.List;

import java.net.URL;
import java.net.MalformedURLException;

import java.io.File;

import org.elasticsearch.ExceptionsHelper;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.Requests;
import org.elasticsearch.client.action.bulk.BulkRequestBuilder;
import org.elasticsearch.cluster.block.ClusterBlockException;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.support.XContentMapValues;
import org.elasticsearch.indices.IndexAlreadyExistsException;
import org.elasticsearch.river.AbstractRiverComponent;
import org.elasticsearch.river.River;
import org.elasticsearch.river.RiverName;
import org.elasticsearch.river.RiverSettings;
import org.elasticsearch.threadpool.ThreadPool;

import org.jruby.Ruby;
import org.jruby.RubyModule;
import org.jruby.javasupport.JavaUtil;
import org.jruby.RubyClass;
import org.jruby.embed.ScriptingContainer;
import org.jruby.embed.PathType;
import org.jruby.embed.AttributeName;
import org.jruby.javasupport.util.RuntimeHelpers;
import org.jruby.runtime.builtin.IRubyObject;
import org.jruby.runtime.ThreadContext;
import org.jruby.runtime.GlobalVariable;
import org.jruby.anno.JRubyMethod;
import static org.jruby.runtime.Visibility.*;
import static org.jruby.CompatVersion.*;
import org.jruby.embed.LocalContextScope;

/**
 * @author <NAME> (<EMAIL>)
 */
public class JRubyRiver extends AbstractRiverComponent implements River {
    private ScriptingContainer container;
    private IRubyObject river;
    
    @Inject public JRubyRiver(RiverName riverName, RiverSettings settings, Client client, ThreadPool threadPool) throws Exception {        
        super(riverName, settings);
        this.container = new ScriptingContainer(LocalContextScope.SINGLETHREAD);

        if (settings.settings().containsKey("jruby")) {
            Map<String, Object> riverSettings = (Map<String, Object>) settings.settings().get("jruby");
            
            //String className       = XContentMapValues.nodeStringValue(riverSettings.get("ruby_class"), "River");
            String scriptName      = XContentMapValues.nodeStringValue(riverSettings.get("script_name"), "lib/river.rb");
            String scriptDirectory = XContentMapValues.nodeStringValue(riverSettings.get("script_directory"), null);
            
            if (null != scriptDirectory) {
                logger.info("changing for directory [{}]", scriptDirectory);
                container.getLoadPaths().add(scriptDirectory);
                container.setAttribute(AttributeName.BASE_DIR, scriptDirectory);
                container.setCurrentDirectory(scriptDirectory);
            }
            
            Ruby runtime = getRuntime();
            runtime.defineVariable(new GlobalVariable(runtime, "$river", runtime.getNil()));
            RubyModule kernel = runtime.getKernel();
            kernel.defineAnnotatedMethods(JRubyRiver.class);
            
            List<String> loadPath = (List<String>)riverSettings.get("load_path");
            
            if (loadPath != null) {
                for (String path : loadPath) {
                    logger.info(path);
                    container.getLoadPaths().add(path);
                } 
            }

            logger.info("running script [{}]", scriptName);
            container.runScriptlet(PathType.RELATIVE, scriptName);
            logger.info("found river class [{}]", river());
            
            RubyClass klass = river();
            IRubyObject clientInstance = JavaUtil.convertJavaToRuby(runtime, client);
            IRubyObject threadPoolInstance = JavaUtil.convertJavaToRuby(runtime, threadPool);
            IRubyObject settingsInstance = JavaUtil.convertJavaToRuby(runtime, settings);
            IRubyObject nameInstance = JavaUtil.convertJavaToRuby(runtime, riverName);
            IRubyObject loggerInstance = JavaUtil.convertJavaToRuby(runtime, logger);
            
            this.river = RuntimeHelpers.invoke(runtime.getCurrentContext(), klass, "new", nameInstance, settingsInstance, clientInstance, threadPoolInstance, loggerInstance);
        } else {
            throw new Exception("No options for jruby river found");
        }
    }

    @Override public void start() {
        try {
            RuntimeHelpers.invoke(getRuntime().getCurrentContext(), this.river, "start");
        } catch (Exception e) {
            logger.error("error in River#start [{}]", e);
        }
    }

    @Override public void close() {
        try {
          RuntimeHelpers.invoke(getRuntime().getCurrentContext(), this.river, "close");
        } catch (Exception e) {
          logger.error("error in River#close [{}]", e);
        }
    }
    
    private Ruby getRuntime() {
        return container.getProvider().getRuntime();
    }
    
    private RubyClass river() {
        return (RubyClass)getRuntime().getGlobalVariables().get("$river");
    }
    
    @JRubyMethod(name = "river", required = 1, visibility = PRIVATE)
    public static void river(ThreadContext context, IRubyObject recv, IRubyObject object) {
        context.getRuntime().getGlobalVariables().set("$river", object);
    }

{'fraction_non_alphanumeric': 0.09959489413743025, 'fraction_numerical': 0.0007643506840938623, 'mean_word_length': 3.2397926117952043, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, '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': '3140693', 'n_tokens_mistral': 3422, 'n_tokens_neox': 3361, 'n_words': 813}