You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@metron.apache.org by ni...@apache.org on 2018/01/12 18:59:47 UTC
[3/4] metron git commit: METRON-1382 Run Stellar in a Zeppelin
Notebook (nickwallen) closes apache/metron#884
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarExecutor.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarExecutor.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarExecutor.java
deleted file mode 100644
index 1be38c3..0000000
--- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarExecutor.java
+++ /dev/null
@@ -1,327 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell;
-
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Maps;
-import java.io.ByteArrayInputStream;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.SortedMap;
-import java.util.concurrent.locks.ReadWriteLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-import org.apache.commons.collections4.trie.PatriciaTrie;
-import org.apache.commons.lang.StringUtils;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.metron.stellar.common.StellarProcessor;
-import org.apache.metron.stellar.common.configuration.ConfigurationsUtils;
-import org.apache.metron.stellar.common.utils.JSONUtils;
-import org.apache.metron.stellar.dsl.Context;
-import org.apache.metron.stellar.dsl.MapVariableResolver;
-import org.apache.metron.stellar.dsl.StellarFunctionInfo;
-import org.apache.metron.stellar.dsl.StellarFunctions;
-import org.apache.metron.stellar.dsl.VariableResolver;
-import org.apache.metron.stellar.dsl.functions.resolver.FunctionResolver;
-import org.jboss.aesh.console.Console;
-
-import static org.apache.metron.stellar.common.configuration.ConfigurationsUtils.readGlobalConfigBytesFromZookeeper;
-import static org.apache.metron.stellar.common.shell.StellarExecutor.OperationType.DOC;
-import static org.apache.metron.stellar.common.shell.StellarExecutor.OperationType.NORMAL;
-import static org.apache.metron.stellar.dsl.Context.Capabilities.GLOBAL_CONFIG;
-import static org.apache.metron.stellar.dsl.Context.Capabilities.STELLAR_CONFIG;
-import static org.apache.metron.stellar.dsl.Context.Capabilities.ZOOKEEPER_CLIENT;
-
-/**
- * Executes Stellar expressions and maintains state across multiple invocations.
- */
-public class StellarExecutor {
-
- public static String SHELL_VARIABLES = "shellVariables";
- public static String CONSOLE = "console";
-
- private ReadWriteLock indexLock = new ReentrantReadWriteLock();
-
- public static class VariableResult {
- private String expression;
- private Object result;
-
- public VariableResult(String expression, Object result) {
- this.expression = expression;
- this.result = result;
- }
-
- public String getExpression() {
- return expression;
- }
-
- public Object getResult() {
- return result;
- }
-
- @Override
- public String toString() {
- String ret = "" + result;
- if(expression != null) {
- ret += " via " + expression;
- }
- return ret;
- }
- }
-
- /**
- * Prefix tree index of auto-completes.
- */
- private PatriciaTrie<AutoCompleteType> autocompleteIndex;
-
- /**
- * The variables known by Stellar.
- */
- private Map<String, VariableResult> variables;
-
- /**
- * The function resolver.
- */
- private FunctionResolver functionResolver;
-
- /**
- * A Zookeeper client. Only defined if given a valid Zookeeper URL.
- */
- private Optional<CuratorFramework> client;
-
- /**
- * The Stellar execution context.
- */
- private Context context;
-
- private Console console;
-
- public enum OperationType {
- DOC
- , MAGIC
- , NORMAL
- }
-
- public interface AutoCompleteTransformation {
- String transform(OperationType type, String key);
- }
-
- public enum AutoCompleteType implements AutoCompleteTransformation{
- FUNCTION((type, key) -> {
- if(type == DOC) {
- return StellarShell.DOC_PREFIX + key;
- }
- else if(type == NORMAL) {
- return key + "(";
- }
- return key;
- })
- , VARIABLE((type, key) -> key )
- , TOKEN((type, key) -> key)
- ;
-
- AutoCompleteTransformation transform;
- AutoCompleteType(AutoCompleteTransformation transform) {
- this.transform = transform;
- }
-
- @Override
- public String transform(OperationType type, String key) {
- return transform.transform(type, key);
- }
- }
-
- /**
- * @param console The console used to drive the REPL.
- * @param properties The Stellar properties.
- * @throws Exception
- */
- public StellarExecutor(Console console, Properties properties) throws Exception {
- this(null, console, properties);
- }
-
- /**
- * @param console The console used to drive the REPL.
- * @param properties The Stellar properties.
- * @throws Exception
- */
- public StellarExecutor(String zookeeperUrl, Console console, Properties properties) throws Exception {
- this.variables = new HashMap<>();
- this.client = createClient(zookeeperUrl);
- this.context = createContext(properties);
-
- // initialize the default function resolver
- StellarFunctions.initialize(this.context);
- this.functionResolver = StellarFunctions.FUNCTION_RESOLVER();
-
- this.autocompleteIndex = initializeIndex();
- this.console = console;
-
- // asynchronously update the index with function names found from a classpath scan.
- new Thread( () -> {
- Iterable<StellarFunctionInfo> functions = functionResolver.getFunctionInfo();
- indexLock.writeLock().lock();
- try {
- for(StellarFunctionInfo info: functions) {
- String functionName = info.getName();
- autocompleteIndex.put(functionName, AutoCompleteType.FUNCTION);
- }
- } finally {
- System.out.println("Functions loaded, you may refer to functions now...");
- indexLock.writeLock().unlock();
- }
- }).start();
- }
-
- private PatriciaTrie<AutoCompleteType> initializeIndex() {
- Map<String, AutoCompleteType> index = new HashMap<>();
-
- index.put("==", AutoCompleteType.TOKEN);
- index.put(">=", AutoCompleteType.TOKEN);
- index.put("<=", AutoCompleteType.TOKEN);
- index.put(":=", AutoCompleteType.TOKEN);
- index.put("quit", AutoCompleteType.TOKEN);
- index.put(StellarShell.MAGIC_FUNCTIONS, AutoCompleteType.FUNCTION);
- index.put(StellarShell.MAGIC_VARS, AutoCompleteType.FUNCTION);
- index.put(StellarShell.MAGIC_GLOBALS, AutoCompleteType.FUNCTION);
- index.put(StellarShell.MAGIC_DEFINE, AutoCompleteType.FUNCTION);
- index.put(StellarShell.MAGIC_UNDEFINE, AutoCompleteType.FUNCTION);
- return new PatriciaTrie<>(index);
- }
-
- public Iterable<String> autoComplete(String buffer, final OperationType opType) {
- indexLock.readLock().lock();
- try {
- SortedMap<String, AutoCompleteType> ret = autocompleteIndex.prefixMap(buffer);
- if (ret.isEmpty()) {
- return new ArrayList<>();
- }
- return Iterables.transform(ret.entrySet(), kv -> kv.getValue().transform(opType, kv.getKey()));
- }
- finally {
- indexLock.readLock().unlock();
- }
- }
-
- /**
- * Creates a Zookeeper client.
- * @param zookeeperUrl The Zookeeper URL.
- */
- private Optional<CuratorFramework> createClient(String zookeeperUrl) {
-
- // can only create client, if have valid zookeeper URL
- if(StringUtils.isNotBlank(zookeeperUrl)) {
- CuratorFramework client = ConfigurationsUtils.getClient(zookeeperUrl);
- client.start();
- return Optional.of(client);
-
- } else {
- return Optional.empty();
- }
- }
-
- /**
- * Creates a Context initialized with configuration stored in Zookeeper.
- */
- private Context createContext(Properties properties) throws Exception {
-
- Context.Builder contextBuilder = new Context.Builder()
- .with(SHELL_VARIABLES, () -> variables)
- .with(CONSOLE, () -> console)
- .with(STELLAR_CONFIG, () -> properties);
-
- // load global configuration from zookeeper
- if (client.isPresent()) {
-
- // fetch the global configuration
- Map<String, Object> global = JSONUtils.INSTANCE.load(
- new ByteArrayInputStream(readGlobalConfigBytesFromZookeeper(client.get())),
- new TypeReference<Map<String, Object>>() {});
-
- contextBuilder
- .with(GLOBAL_CONFIG, () -> global)
- .with(ZOOKEEPER_CLIENT, () -> client.get())
- .with(STELLAR_CONFIG, () -> getStellarConfig(global, properties));
- }
-
- return contextBuilder.build();
- }
-
- private Map<String, Object> getStellarConfig(Map<String, Object> globalConfig, Properties props) {
- Map<String, Object> ret = new HashMap<>();
- ret.putAll(globalConfig);
- if(props != null) {
- for (Map.Entry<Object, Object> kv : props.entrySet()) {
- ret.put(kv.getKey().toString(), kv.getValue());
- }
- }
- return ret;
- }
-
- /**
- * Executes the Stellar expression and returns the result.
- * @param expression The Stellar expression to execute.
- * @return The result of the expression.
- */
- public Object execute(String expression) {
- VariableResolver variableResolver = new MapVariableResolver(Maps.transformValues(variables, result -> result.getResult())
- , Collections.emptyMap());
- StellarProcessor processor = new StellarProcessor();
- return processor.parse(expression, variableResolver, functionResolver, context);
- }
-
- /**
- * Assigns a value to a variable.
- * @param variable The name of the variable.
- * @param value The value of the variable
- */
- public void assign(String variable, String expression, Object value) {
- this.variables.put(variable, new VariableResult(expression, value));
- indexLock.writeLock().lock();
- try {
- if (value != null) {
- this.autocompleteIndex.put(variable, AutoCompleteType.VARIABLE);
- } else {
- this.autocompleteIndex.remove(variable);
- }
- }
- finally {
- indexLock.writeLock().unlock();
- }
- }
-
- public Map<String, VariableResult> getVariables() {
- return this.variables;
- }
-
- public FunctionResolver getFunctionResolver() {
- return functionResolver;
- }
-
- public Context getContext() {
- return context;
- }
-}
-
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarResult.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarResult.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarResult.java
new file mode 100644
index 0000000..4e5c81e
--- /dev/null
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarResult.java
@@ -0,0 +1,195 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell;
+
+import java.util.Optional;
+
+/**
+ * The result of executing a Stellar expression within a StellarShellExecutor.
+ */
+public class StellarResult {
+
+ /**
+ * Indicates that a Stellar expression resulted in either
+ * success or an error.
+ */
+ enum Status {
+ SUCCESS,
+ ERROR,
+ TERMINATE
+ }
+
+ /**
+ * Indicates either success or failure of executing the expression.
+ */
+ private Status status;
+
+ /**
+ * The result of executing the expression. Only valid when execution is successful.
+ */
+ private Optional<Object> value;
+
+ /**
+ * The error that occurred when executing the expression. Only valid when execution results in an error.
+ */
+ private Optional<Throwable> exception;
+
+ /**
+ * Indicates if the value is null;
+ *
+ * A null is a valid result, but cannot be unwrapped from an Optional. Because of this
+ * a boolean is used to indicate if the result is a success and the value is null.
+ */
+ private boolean isValueNull;
+
+ /**
+ * Private constructor to construct a result indicate success. Use the static methods; success.
+ *
+ * @param status Indicates success or failure.
+ * @param value The value of executing the expression.
+ */
+ private StellarResult(Status status, Object value) {
+ this.status = status;
+ this.value = Optional.ofNullable(value);
+ this.exception = Optional.empty();
+ this.isValueNull = (value == null) && (status == Status.SUCCESS);
+ }
+
+ /**
+ * Private constructor to construct a result indicating an error occurred. Use the static method; error.
+ *
+ * @param status Indicates success or failure.
+ * @param exception The exception that occurred when executing the expression.
+ */
+ private StellarResult(Status status, Throwable exception) {
+ this.status = status;
+ this.value = Optional.empty();
+ this.exception = Optional.of(exception);
+ this.isValueNull = false;
+ }
+
+ /**
+ * Create a result indicating the execution of an expression was successful.
+ *
+ * @param value The result of executing the expression.
+ * @return A Result indicating success.
+ */
+ public static StellarResult success(Object value) {
+ return new StellarResult(Status.SUCCESS, value);
+ }
+
+ /**
+ * Create a result indicating that the execution of an expression was not successful.
+ *
+ * @param exception The exception that occurred while executing the expression.
+ * @return A Result indicating that an error occurred.
+ */
+ public static StellarResult error(Throwable exception) {
+ return new StellarResult(Status.ERROR, exception);
+ }
+
+ /**
+ * Create a result indicating that the execution of an expression was not successful.
+ *
+ * @param errorMessage An error message.
+ * @return A Result indicating that an error occurred.
+ */
+ public static StellarResult error(String errorMessage) {
+ return new StellarResult(Status.ERROR, new IllegalArgumentException(errorMessage));
+ }
+
+ /**
+ * Indicates an empty result; one that is successful yet has no result. For example,
+ * executing a comment.
+ *
+ * @return An empty result.
+ */
+ public static StellarResult noop() {
+ return new StellarResult(Status.SUCCESS, "");
+ }
+
+ /**
+ * Indicates that the user would like to terminate the session.
+ *
+ * @return A result indicating that the session should be terminated.
+ */
+ public static StellarResult terminate() {
+ return new StellarResult(Status.TERMINATE, "");
+ }
+
+ /**
+ * @return True, if the result indicates success. Otherwise, false.
+ */
+ public boolean isSuccess() {
+ return status == Status.SUCCESS;
+ }
+
+ /**
+ * @return True, if the result indicates an error. Otherwise, false.
+ */
+ public boolean isError() {
+ return status == Status.ERROR;
+ }
+
+ /**
+ * @return True, if status indicates terminate was requested. Otherwise, false.
+ */
+ public boolean isTerminate() {
+ return status == Status.TERMINATE;
+ }
+
+ /**
+ * @return True, if the value is null. Otherwise, false.
+ */
+ public boolean isValueNull() {
+ return isValueNull;
+ }
+
+ /**
+ * @return The status which indicates success or failure.
+ */
+ public Status getStatus() {
+ return status;
+ }
+
+ /**
+ * @return An optional value that only applies when status is success.
+ */
+ public Optional<Object> getValue() {
+ return value;
+ }
+
+ /**
+ * @return An optional exception that only applies when status is error.
+ */
+ public Optional<Throwable> getException() {
+ return exception;
+ }
+
+ @Override
+ public String toString() {
+ return "StellarResult{" +
+ "status=" + status +
+ ", value=" + value +
+ ", exception=" + exception +
+ ", isValueNull=" + isValueNull +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShell.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShell.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShell.java
deleted file mode 100644
index 6f23a5e..0000000
--- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShell.java
+++ /dev/null
@@ -1,557 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell;
-
-import com.fasterxml.jackson.core.type.TypeReference;
-import com.google.common.base.Splitter;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import org.apache.commons.cli.*;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.log4j.PropertyConfigurator;
-import org.apache.metron.stellar.common.StellarAssignment;
-import org.apache.metron.stellar.common.utils.JSONUtils;
-import org.apache.metron.stellar.dsl.StellarFunctionInfo;
-import org.jboss.aesh.complete.CompleteOperation;
-import org.jboss.aesh.complete.Completion;
-import org.jboss.aesh.console.AeshConsoleCallback;
-import org.jboss.aesh.console.Console;
-import org.jboss.aesh.console.ConsoleOperation;
-import org.jboss.aesh.console.Prompt;
-import org.jboss.aesh.console.settings.SettingsBuilder;
-import org.jboss.aesh.terminal.CharacterType;
-import org.jboss.aesh.terminal.Color;
-import org.jboss.aesh.terminal.TerminalCharacter;
-import org.jboss.aesh.terminal.TerminalColor;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.function.Predicate;
-import java.util.stream.Collectors;
-import java.util.stream.StreamSupport;
-
-import static org.apache.metron.stellar.dsl.Context.Capabilities.GLOBAL_CONFIG;
-
-/**
- * A REPL environment for Stellar.
- *
- * Useful for debugging Stellar expressions.
- */
-public class StellarShell extends AeshConsoleCallback implements Completion {
-
- private static final String WELCOME = "Stellar, Go!\n" +
- "Please note that functions are loading lazily in the background and will be unavailable until loaded fully.";
- private List<TerminalCharacter> EXPRESSION_PROMPT = new ArrayList<TerminalCharacter>()
- {{
- add(new TerminalCharacter('[', new TerminalColor(Color.RED, Color.DEFAULT)));
- add(new TerminalCharacter('S', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
- add(new TerminalCharacter('t', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
- add(new TerminalCharacter('e', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
- add(new TerminalCharacter('l', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
- add(new TerminalCharacter('l', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
- add(new TerminalCharacter('a', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
- add(new TerminalCharacter('r', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
- add(new TerminalCharacter(']', new TerminalColor(Color.RED, Color.DEFAULT)));
- add(new TerminalCharacter('>', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.UNDERLINE));
- add(new TerminalCharacter('>', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.UNDERLINE));
- add(new TerminalCharacter('>', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.UNDERLINE));
- add(new TerminalCharacter(' ', new TerminalColor(Color.DEFAULT, Color.DEFAULT)));
- }};
-
- public static final String ERROR_PROMPT = "[!] ";
- public static final String MAGIC_PREFIX = "%";
- public static final String MAGIC_FUNCTIONS = MAGIC_PREFIX + "functions";
- public static final String MAGIC_VARS = MAGIC_PREFIX + "vars";
- public static final String DOC_PREFIX = "?";
- public static final String STELLAR_PROPERTIES_FILENAME = "stellar.properties";
- public static final String MAGIC_GLOBALS = MAGIC_PREFIX + "globals";
- public static final String MAGIC_DEFINE = MAGIC_PREFIX + "define";
- public static final String MAGIC_UNDEFINE = MAGIC_PREFIX + "undefine";
-
- private StellarExecutor executor;
-
- private Console console;
-
- /**
- * Execute the Stellar REPL.
- */
- public static void main(String[] args) throws Exception {
- StellarShell shell = new StellarShell(args);
- shell.execute();
- }
-
- /**
- * Create a Stellar REPL.
- * @param args The commmand-line arguments.
- */
- public StellarShell(String[] args) throws Exception {
-
- // define valid command-line options
- Options options = new Options();
- options.addOption("z", "zookeeper", true, "Zookeeper URL fragment in the form [HOSTNAME|IPADDRESS]:PORT");
- options.addOption("v", "variables", true, "File containing a JSON Map of variables");
- options.addOption("irc", "inputrc", true, "File containing the inputrc if not the default ~/.inputrc");
- options.addOption("na", "no_ansi", false, "Make the input prompt not use ANSI colors.");
- options.addOption("h", "help", false, "Print help");
- options.addOption("p", "properties", true, "File containing Stellar properties");
- {
- Option o = new Option("l", "log4j", true, "The log4j properties file to load");
- o.setArgName("FILE");
- o.setRequired(false);
- options.addOption(o);
- }
- CommandLineParser parser = new PosixParser();
- CommandLine commandLine = parser.parse(options, args);
-
- // print help
- if(commandLine.hasOption("h")) {
- HelpFormatter formatter = new HelpFormatter();
- formatter.printHelp("stellar", options);
- System.exit(0);
- }
-
- try {
- StellarShellOptionsValidator.validateOptions(commandLine);
- }catch(IllegalArgumentException e){
- System.out.println(e.getMessage());
- System.exit(1);
- }
- //setting up logging if specified
- if(commandLine.hasOption("l")) {
- PropertyConfigurator.configure(commandLine.getOptionValue("l"));
- }
- console = createConsole(commandLine);
- executor = createExecutor(commandLine, console, getStellarProperties(commandLine));
- loadVariables(commandLine, executor);
- console.setPrompt(new Prompt(EXPRESSION_PROMPT));
- console.addCompletion(this);
- console.setConsoleCallback(this);
- }
-
- /**
- * Loads any variables defined in an external file.
- * @param commandLine The command line arguments.
- * @param executor The stellar executor.
- * @throws IOException
- */
- private static void loadVariables(CommandLine commandLine, StellarExecutor executor) throws IOException {
- if(commandLine.hasOption("v")) {
-
- Map<String, Object> variables = JSONUtils.INSTANCE.load(
- new File(commandLine.getOptionValue("v")),
- new TypeReference<Map<String, Object>>() {});
-
- for(Map.Entry<String, Object> kv : variables.entrySet()) {
- executor.assign(kv.getKey(), null, kv.getValue());
- }
- }
- }
-
- /**
- * Creates the Stellar execution environment.
- * @param commandLine The command line arguments.
- * @param console The console which drives the REPL.
- * @param properties Stellar properties.
- */
- private static StellarExecutor createExecutor(CommandLine commandLine, Console console, Properties properties) throws Exception {
- StellarExecutor executor;
-
- // create the executor
- if(commandLine.hasOption("z")) {
- String zookeeperUrl = commandLine.getOptionValue("z");
- executor = new StellarExecutor(zookeeperUrl, console, properties);
-
- } else {
- executor = new StellarExecutor(console, properties);
- }
-
- return executor;
- }
-
- /**
- * Creates the REPL's console.
- * @param commandLine The command line options.
- */
- private Console createConsole(CommandLine commandLine) {
-
- // console settings
- boolean useAnsi = !commandLine.hasOption("na");
- SettingsBuilder settings = new SettingsBuilder().enableAlias(true)
- .enableMan(true)
- .ansi(useAnsi)
- .parseOperators(false)
- .inputStream(PausableInput.INSTANCE);
-
- if(commandLine.hasOption("irc")) {
- settings = settings.inputrc(new File(commandLine.getOptionValue("irc")));
- }
-
- return new Console(settings.create());
- }
-
- /**
- * Retrieves the Stellar properties. The properties are either loaded from a file in
- * the classpath or a set of defaults are used.
- */
- private Properties getStellarProperties(CommandLine commandLine) throws IOException {
- Properties properties = new Properties();
-
- if (commandLine.hasOption("p")) {
-
- // first attempt to load properties from a file specified on the command-line
- try (InputStream in = new FileInputStream(commandLine.getOptionValue("p"))) {
- if(in != null) {
- properties.load(in);
- }
- }
-
- } else {
-
- // otherwise attempt to load properties from the classpath
- try (InputStream in = getClass().getClassLoader().getResourceAsStream(STELLAR_PROPERTIES_FILENAME)) {
- if(in != null) {
- properties.load(in);
- }
- }
- }
-
- return properties;
- }
-
- /**
- * Handles the main loop for the REPL.
- */
- public void execute() {
-
- // welcome message and print globals
- writeLine(WELCOME);
- executor.getContext()
- .getCapability(GLOBAL_CONFIG, false)
- .ifPresent(conf -> writeLine(conf.toString()));
-
- console.start();
- }
-
- /**
- * Handles user interaction when executing a Stellar expression.
- * @param expression The expression to execute.
- */
- private void handleStellar(String expression) {
-
- String stellarExpression = expression;
- String variable = null;
- if(StellarAssignment.isAssignment(expression)) {
- StellarAssignment expr = StellarAssignment.from(expression);
- variable = expr.getVariable();
- stellarExpression = expr.getStatement();
- }
- else {
- if (!stellarExpression.isEmpty()) {
- stellarExpression = stellarExpression.trim();
- }
- }
-
- try {
- Object result = executor.execute(stellarExpression);
- if (result != null && variable == null) {
- writeLine(result.toString());
- }
- if (variable != null) {
- executor.assign(variable, stellarExpression, result);
- }
- } catch (Throwable t) {
- if(variable != null) {
- writeLine(String.format("%s ERROR: Variable %s not assigned", ERROR_PROMPT, variable));
- }
- writeLine(ERROR_PROMPT + t.getMessage());
- t.printStackTrace();
- }
- }
-
- /**
- * Executes a magic expression.
- * @param rawExpression The expression to execute.
- */
- private void handleMagic(String rawExpression) {
-
- String[] expression = rawExpression.trim().split("\\s+");
- String command = expression[0];
-
- if (MAGIC_FUNCTIONS.equals(command)) {
- handleMagicFunctions(expression);
-
- } else if (MAGIC_VARS.equals(command)) {
- handleMagicVars();
-
- } else if (MAGIC_GLOBALS.equals(command)) {
- handleMagicGlobals();
-
- } else if (MAGIC_DEFINE.equals(command)) {
- handleMagicDefine(rawExpression);
-
- } else if(MAGIC_UNDEFINE.equals(command)) {
- handleMagicUndefine(expression);
-
- } else {
- writeLine(ERROR_PROMPT + "undefined magic command: " + rawExpression);
- }
- }
-
- /**
- * Handle a magic '%functions'. Lists all of the variables in-scope.
- * @param expression
- */
- private void handleMagicFunctions(String[] expression) {
-
- // if '%functions FOO' then show only functions that contain 'FOO'
- Predicate<String> nameFilter = (name -> true);
- if (expression.length > 1) {
- nameFilter = (name -> name.contains(expression[1]));
- }
-
- // '%functions' -> list all functions in scope
- String functions = StreamSupport
- .stream(executor.getFunctionResolver().getFunctionInfo().spliterator(), false)
- .map(info -> String.format("%s", info.getName()))
- .filter(nameFilter)
- .sorted()
- .collect(Collectors.joining(", "));
- writeLine(functions);
- }
-
- /**
- * Handle a magic '%vars'. Lists all of the variables in-scope.
- */
- private void handleMagicVars() {
- executor.getVariables()
- .forEach((k, v) -> writeLine(String.format("%s = %s", k, v)));
- }
-
- /**
- * Handle a magic '%globals'. List all of the global configuration values.
- */
- private void handleMagicGlobals() {
- Map<String, Object> globals = getOrCreateGlobalConfig(executor);
- writeLine(globals.toString());
- }
-
- /**
- * Handle a magic '%define var=value'. Alter the global configuration.
- * @param expression The expression passed to %define
- */
- public void handleMagicDefine(String expression) {
-
- // grab the expression in '%define <assign-expression>'
- String assignExpr = StringUtils.trimToEmpty(expression.substring(MAGIC_DEFINE.length()));
- if (assignExpr.length() > 0) {
-
- // the expression must be an assignment
- if(StellarAssignment.isAssignment(assignExpr)) {
- StellarAssignment expr = StellarAssignment.from(assignExpr);
-
- // execute the expression
- Object result = executor.execute(expr.getStatement());
- if (result != null) {
- writeLine(result.toString());
-
- // alter the global configuration
- getOrCreateGlobalConfig(executor).put(expr.getVariable(), result);
- }
-
- } else {
- // the expression is not an assignment. boo!
- writeLine(ERROR_PROMPT + MAGIC_DEFINE + " expected assignment expression");
- }
- }
- }
-
- /**
- * Handle a magic '%undefine var'. Removes a variable from the global configuration.
- * @param expression
- */
- private void handleMagicUndefine(String[] expression) {
- if(expression.length > 1) {
- Map<String, Object> globals = getOrCreateGlobalConfig(executor);
- globals.remove(expression[1]);
- }
- }
-
- /**
- * Retrieves the GLOBAL_CONFIG, if it exists. If it does not, it creates the GLOBAL_CONFIG
- * and adds it to the Stellar execution context.
- * @param executor The Stellar executor.
- * @return The global configuration.
- */
- private Map<String, Object> getOrCreateGlobalConfig(StellarExecutor executor) {
- Map<String, Object> globals;
- Optional<Object> capability = executor.getContext().getCapability(GLOBAL_CONFIG, false);
- if (capability.isPresent()) {
- globals = (Map<String, Object>) capability.get();
-
- } else {
- // if it does not exist, create it. this creates the global config for the current stellar executor
- // session only. this does not change the global config maintained externally in zookeeper
- globals = new HashMap<>();
- executor.getContext().addCapability(GLOBAL_CONFIG, () -> globals);
- }
- return globals;
- }
-
- /**
- * Executes a doc expression.
- * @param expression The doc expression to execute.
- */
- private void handleDoc(String expression) {
-
- String functionName = StringUtils.substring(expression, 1);
- StreamSupport
- .stream(executor.getFunctionResolver().getFunctionInfo().spliterator(), false)
- .filter(info -> StringUtils.equals(functionName, info.getName()))
- .map(info -> format(info))
- .forEach(doc -> write(doc));
- }
-
- /**
- * Executes a quit.
- */
- private void handleQuit() {
- try {
- console.stop();
- } catch (Throwable e) {
- e.printStackTrace();
- }
- }
-
- /**
- * Formats the Stellar function info object into a readable string.
- * @param info The stellar function info object.
- * @return A readable string.
- */
- private String format(StellarFunctionInfo info) {
- StringBuffer ret = new StringBuffer();
- ret.append(info.getName() + "\n");
- ret.append(String.format("Description: %-60s\n\n", info.getDescription()));
- if(info.getParams().length > 0) {
- ret.append("Arguments:\n");
- for(String param : info.getParams()) {
- ret.append(String.format("\t%-60s\n", param));
- }
- ret.append("\n");
- }
- ret.append(String.format("Returns: %-60s\n", info.getReturns()));
-
- return ret.toString();
- }
-
- /**
- * Is a given expression a built-in magic?
- * @param expression The expression.
- */
- private boolean isMagic(String expression) {
- return StringUtils.startsWith(expression, MAGIC_PREFIX);
- }
-
- /**
- * Is a given expression asking for function documentation?
- * @param expression The expression.
- */
- private boolean isDoc(String expression) {
- return StringUtils.startsWith(expression, DOC_PREFIX);
- }
-
- private void write(String out) {
- System.out.print(out);
- }
-
- private void writeLine(String out) {
- console.getShell().out().println(out);
- }
-
- @Override
- public int execute(ConsoleOperation output) throws InterruptedException {
- String expression = output.getBuffer().trim();
- if(StringUtils.isNotBlank(expression) ) {
- if(isMagic(expression)) {
- handleMagic( expression);
-
- } else if(isDoc(expression)) {
- handleDoc(expression);
-
- } else if (expression.equals("quit")) {
- handleQuit();
-
- } else if(expression.charAt(0) == '#') {
- return 0; // comment, do nothing
-
- } else {
- handleStellar(expression);
- }
- }
-
- return 0;
- }
-
- @Override
- public void complete(CompleteOperation completeOperation) {
- if(!completeOperation.getBuffer().isEmpty()) {
- String lastToken = Iterables.getLast(Splitter.on(" ").split(completeOperation.getBuffer()), null);
- if(lastToken != null && !lastToken.isEmpty()) {
- lastToken = lastToken.trim();
- final String lastBit = lastToken;
- final boolean isDocRequest = isDoc(lastToken);
- if(isDocRequest) {
- lastToken = lastToken.substring(1);
- }
- StellarExecutor.OperationType opType = StellarExecutor.OperationType.NORMAL;
- if(isDocRequest) {
- opType = StellarExecutor.OperationType.DOC;
- }
- else if(isMagic(lastToken)) {
- opType = StellarExecutor.OperationType.MAGIC;
- }
- Iterable<String> candidates = executor.autoComplete(lastToken, opType);
- if(candidates != null && !Iterables.isEmpty(candidates)) {
- completeOperation.setCompletionCandidates( Lists.newArrayList(
- Iterables.transform(candidates, s -> stripOff(completeOperation.getBuffer(), lastBit) + s )
- )
- );
- }
- }
- }
- }
-
- private static String stripOff(String baseString, String lastBit) {
- int index = baseString.lastIndexOf(lastBit);
- if(index < 0) {
- return baseString;
- }
- return baseString.substring(0, index);
- }
-}
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShellExecutor.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShellExecutor.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShellExecutor.java
new file mode 100644
index 0000000..7e2d4fc
--- /dev/null
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShellExecutor.java
@@ -0,0 +1,79 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell;
+
+import org.apache.metron.stellar.dsl.Context;
+import org.apache.metron.stellar.dsl.functions.resolver.FunctionResolver;
+
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Responsible for executing Stellar in a shell-like environment.
+ *
+ * Provides the additional capabilities expected of executing Stellar
+ * in a shell-like environment including maintaining state, variable assignment,
+ * magic commands, doc strings, and comments.
+ */
+public interface StellarShellExecutor extends StellarExecutionNotifier {
+
+ /**
+ * Initialize the Stellar executor.
+ */
+ void init();
+
+ /**
+ * Execute the Stellar expression.
+ * @param expression The Stellar expression to execute.
+ * @return The result of executing the Stellar expression.
+ */
+ StellarResult execute(String expression);
+
+ /**
+ * Update the state of the executor by assign a value to a variable.
+ * @param variable The name of the variable.
+ * @param value The value to assign.
+ * @param expression The expression that resulted in the given value. Optional.
+ */
+ void assign(String variable, Object value, Optional<String> expression);
+
+ /**
+ * The current state of the Stellar execution environment.
+ */
+ Map<String, VariableResult> getState();
+
+ /**
+ * Returns the Context for the Stellar execution environment.
+ * @return The execution context.
+ */
+ Context getContext();
+
+ /**
+ * Returns the global configuration of the Stellar execution environment.
+ * @return A map of values defined in the global configuration.
+ */
+ Map<String, Object> getGlobalConfig();
+
+ /**
+ * Returns the function resolver of the Stellar execution environment.
+ * @return The function resolver.
+ */
+ FunctionResolver getFunctionResolver();
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShellOptionsValidator.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShellOptionsValidator.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShellOptionsValidator.java
deleted file mode 100644
index ab92401..0000000
--- a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/StellarShellOptionsValidator.java
+++ /dev/null
@@ -1,127 +0,0 @@
-/*
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell;
-
-import java.io.File;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.function.Predicate;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import com.google.common.base.Splitter;
-import org.apache.commons.cli.CommandLine;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.validator.routines.InetAddressValidator;
-
-public class StellarShellOptionsValidator {
-
- private static final Pattern validPortPattern = Pattern.compile("(^.*)[:](\\d+)$");
- private static final Predicate<String> hostnameValidator = hostname -> {
- if(StringUtils.isEmpty(hostname)) {
- return false;
- }
- try {
- InetAddress add = InetAddress.getByName(hostname);
- return true;
- } catch (UnknownHostException e) {
- return false;
- }
- };
-
-
-
- private static final InetAddressValidator inetAddressValidator = InetAddressValidator
- .getInstance();
-
- /**
- * Validates Stellar CLI Options.
- */
- public static void validateOptions(CommandLine commandLine) throws IllegalArgumentException {
- if (commandLine.hasOption('z')) {
- validateZookeeperOption(commandLine.getOptionValue('z'));
- }
- // v, irc, p are files
- if (commandLine.hasOption('v')) {
- validateFileOption("v", commandLine.getOptionValue('v'));
- }
- if (commandLine.hasOption("irc")) {
- validateFileOption("irc", commandLine.getOptionValue("irc"));
- }
- if (commandLine.hasOption('p')) {
- validateFileOption("p", commandLine.getOptionValue('p'));
- }
-
- }
-
- /**
- * Zookeeper argument should be in the form [HOST|IP]:PORT.
- *
- * @param zMulti the zookeeper url fragment
- */
- private static void validateZookeeperOption(String zMulti) throws IllegalArgumentException {
- for(String z : Splitter.on(",").split(zMulti)) {
- Matcher matcher = validPortPattern.matcher(z);
- boolean hasPort = z.contains(":");
- if (hasPort && !matcher.matches()) {
- throw new IllegalArgumentException(String.format("Zookeeper option must have valid port: %s", z));
- }
-
- if (hasPort && matcher.groupCount() != 2) {
- throw new IllegalArgumentException(
- String.format("Zookeeper Option must be in the form of [HOST|IP]:PORT %s", z));
- }
- String name = hasPort?matcher.group(1):z;
- Integer port = hasPort?Integer.parseInt(matcher.group(2)):null;
-
- if (!hostnameValidator.test(name) && !inetAddressValidator.isValid(name)) {
- throw new IllegalArgumentException(
- String.format("Zookeeper Option %s is not a valid host name or ip address %s", name, z));
- }
-
- if (hasPort && (port == 0 || port > 65535)) {
- throw new IllegalArgumentException(
- String.format("Zookeeper Option %s port is not valid", z));
- }
- }
- }
-
- /**
- * File options must exist and be readable.
- *
- * @param option name of the option
- * @param fileName the file name
- */
- private static void validateFileOption(String option, String fileName)
- throws IllegalArgumentException {
- File file = new File(fileName);
- if (!file.exists()) {
- throw new IllegalArgumentException(
- String.format("%s: File %s doesn't exist", option, fileName));
- }
- if (!file.canRead()) {
- throw new IllegalArgumentException(
- String.format("%s: File %s is not readable", option, fileName));
- }
- }
-}
-
-
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/VariableResult.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/VariableResult.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/VariableResult.java
new file mode 100644
index 0000000..6cb1173
--- /dev/null
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/VariableResult.java
@@ -0,0 +1,101 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell;
+
+import java.util.Optional;
+
+/**
+ * The value assigned to a variable.
+ *
+ * Allows us to maintain not only the resulting value, but the
+ * expression that resulted in that value.
+ */
+public class VariableResult {
+
+ /**
+ * The expression that resulted in the value. Not always available.
+ */
+ private Optional<String> expression;
+
+ /**
+ * The value of the variable.
+ */
+ private Object result;
+
+ /**
+ * Create a new VariableResult when the expression that resulted in a value is known.
+ *
+ * @param value The value.
+ * @param expression The expression that resulted in the given value.
+ * @return A VariableResult.
+ */
+ public static VariableResult withExpression(Object value, String expression) {
+ return new VariableResult(Optional.of(expression), value);
+ }
+
+ /**
+ * Create a new VariableResult when the expression that resulted in a value is known.
+ *
+ * @param value The value.
+ * @param expression The expression that resulted in the given value.
+ * @return A VariableResult.
+ */
+ public static VariableResult withExpression(Object value, Optional<String> expression) {
+ return new VariableResult(expression, value);
+ }
+
+ /**
+ * Create a new VariableResult when only the value is known.
+ *
+ * @param value The value.
+ * @return A VariableResult.
+ */
+ public static VariableResult withValue(Object value) {
+ return new VariableResult(Optional.empty(), value);
+ }
+
+ /**
+ * Private constructor. Use the static method 'withExpression' and 'withValue'.
+ *
+ * @param expression The expression that resulted in the given value.
+ * @param result The value assigned to the variable.
+ */
+ private VariableResult(Optional<String> expression, Object result) {
+ this.expression = expression;
+ this.result = result;
+ }
+
+ public Optional<String> getExpression() {
+ return expression;
+ }
+
+ public Object getResult() {
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ String ret = "" + result;
+ if(getExpression().isPresent()) {
+ ret += " via " + expression.get();
+ }
+ return ret;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/PausableInput.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/PausableInput.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/PausableInput.java
new file mode 100644
index 0000000..fad0115
--- /dev/null
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/PausableInput.java
@@ -0,0 +1,372 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell.cli;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * An input stream which mirrors System.in, but allows you to 'pause' and 'unpause' it.
+ * The Aeshell has an external thread which is constantly polling System.in. If you
+ * need to spawn a program externally (i.e. an editor) which shares stdin, this thread
+ * and the spawned program both share a buffer. This causes contention and unpredictable
+ * results (e.g. an input may be consumed by either the aeshell thread or the spawned program)
+ *
+ * Because you can inject an input stream into the console, we create this which can act as a
+ * facade to System.in under normal 'unpaused' circumstances, and when paused, turn off the
+ * access to System.in. This allows us to turn off access to aeshell while maintaining access
+ * to the external program.
+ *
+ */
+public class PausableInput extends InputStream {
+ InputStream in = System.in;
+ boolean paused = false;
+ private PausableInput() {
+ super();
+ }
+
+ /**
+ * Stop mirroring stdin
+ */
+ public void pause() {
+ paused = true;
+ }
+
+ /**
+ * Resume mirroring stdin.
+ * @throws IOException
+ */
+ public void unpause() throws IOException {
+ in.read(new byte[in.available()]);
+ paused = false;
+ }
+
+ public final static PausableInput INSTANCE = new PausableInput();
+
+ /**
+ * Reads the next byte of data from the input stream. The value byte is
+ * returned as an <code>int</code> in the range <code>0</code> to
+ * <code>255</code>. If no byte is available because the end of the stream
+ * has been reached, the value <code>-1</code> is returned. This method
+ * blocks until input data is available, the end of the stream is detected,
+ * or an exception is thrown.
+ * <p>
+ * <p> A subclass must provide an implementation of this method.
+ *
+ * @return the next byte of data, or <code>-1</code> if the end of the
+ * stream is reached.
+ * @throws IOException if an I/O error occurs.
+ */
+ @Override
+ public int read() throws IOException {
+
+ return in.read();
+ }
+
+ /**
+ * Reads some number of bytes from the input stream and stores them into
+ * the buffer array <code>b</code>. The number of bytes actually read is
+ * returned as an integer. This method blocks until input data is
+ * available, end of file is detected, or an exception is thrown.
+ * <p>
+ * <p> If the length of <code>b</code> is zero, then no bytes are read and
+ * <code>0</code> is returned; otherwise, there is an attempt to read at
+ * least one byte. If no byte is available because the stream is at the
+ * end of the file, the value <code>-1</code> is returned; otherwise, at
+ * least one byte is read and stored into <code>b</code>.
+ * <p>
+ * <p> The first byte read is stored into element <code>b[0]</code>, the
+ * next one into <code>b[1]</code>, and so on. The number of bytes read is,
+ * at most, equal to the length of <code>b</code>. Let <i>k</i> be the
+ * number of bytes actually read; these bytes will be stored in elements
+ * <code>b[0]</code> through <code>b[</code><i>k</i><code>-1]</code>,
+ * leaving elements <code>b[</code><i>k</i><code>]</code> through
+ * <code>b[b.length-1]</code> unaffected.
+ * <p>
+ * <p> The <code>read(b)</code> method for class <code>InputStream</code>
+ * has the same effect as: <pre><code> read(b, 0, b.length) </code></pre>
+ *
+ * @param b the buffer into which the data is read.
+ * @return the total number of bytes read into the buffer, or
+ * <code>-1</code> if there is no more data because the end of
+ * the stream has been reached.
+ * @throws IOException If the first byte cannot be read for any reason
+ * other than the end of the file, if the input stream has been closed, or
+ * if some other I/O error occurs.
+ * @throws NullPointerException if <code>b</code> is <code>null</code>.
+ * @see InputStream#read(byte[], int, int)
+ */
+ @Override
+ public int read(byte[] b) throws IOException {
+
+ if(paused) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return 0;
+ }
+ int ret = in.read(b);
+ return ret;
+ }
+
+ /**
+ * Reads up to <code>len</code> bytes of data from the input stream into
+ * an array of bytes. An attempt is made to read as many as
+ * <code>len</code> bytes, but a smaller number may be read.
+ * The number of bytes actually read is returned as an integer.
+ * <p>
+ * <p> This method blocks until input data is available, end of file is
+ * detected, or an exception is thrown.
+ * <p>
+ * <p> If <code>len</code> is zero, then no bytes are read and
+ * <code>0</code> is returned; otherwise, there is an attempt to read at
+ * least one byte. If no byte is available because the stream is at end of
+ * file, the value <code>-1</code> is returned; otherwise, at least one
+ * byte is read and stored into <code>b</code>.
+ * <p>
+ * <p> The first byte read is stored into element <code>b[off]</code>, the
+ * next one into <code>b[off+1]</code>, and so on. The number of bytes read
+ * is, at most, equal to <code>len</code>. Let <i>k</i> be the number of
+ * bytes actually read; these bytes will be stored in elements
+ * <code>b[off]</code> through <code>b[off+</code><i>k</i><code>-1]</code>,
+ * leaving elements <code>b[off+</code><i>k</i><code>]</code> through
+ * <code>b[off+len-1]</code> unaffected.
+ * <p>
+ * <p> In every case, elements <code>b[0]</code> through
+ * <code>b[off]</code> and elements <code>b[off+len]</code> through
+ * <code>b[b.length-1]</code> are unaffected.
+ * <p>
+ * <p> The <code>read(b,</code> <code>off,</code> <code>len)</code> method
+ * for class <code>InputStream</code> simply calls the method
+ * <code>read()</code> repeatedly. If the first such call results in an
+ * <code>IOException</code>, that exception is returned from the call to
+ * the <code>read(b,</code> <code>off,</code> <code>len)</code> method. If
+ * any subsequent call to <code>read()</code> results in a
+ * <code>IOException</code>, the exception is caught and treated as if it
+ * were end of file; the bytes read up to that point are stored into
+ * <code>b</code> and the number of bytes read before the exception
+ * occurred is returned. The default implementation of this method blocks
+ * until the requested amount of input data <code>len</code> has been read,
+ * end of file is detected, or an exception is thrown. Subclasses are encouraged
+ * to provide a more efficient implementation of this method.
+ *
+ * @param b the buffer into which the data is read.
+ * @param off the start offset in array <code>b</code>
+ * at which the data is written.
+ * @param len the maximum number of bytes to read.
+ * @return the total number of bytes read into the buffer, or
+ * <code>-1</code> if there is no more data because the end of
+ * the stream has been reached.
+ * @throws IOException If the first byte cannot be read for any reason
+ * other than end of file, or if the input stream has been closed, or if
+ * some other I/O error occurs.
+ * @throws NullPointerException If <code>b</code> is <code>null</code>.
+ * @throws IndexOutOfBoundsException If <code>off</code> is negative,
+ * <code>len</code> is negative, or <code>len</code> is greater than
+ * <code>b.length - off</code>
+ * @see InputStream#read()
+ */
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if(paused) {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ return 0;
+ }
+ int ret = in.read(b, off, len);
+ return ret;
+ }
+
+ /**
+ * Skips over and discards <code>n</code> bytes of data from this input
+ * stream. The <code>skip</code> method may, for a variety of reasons, end
+ * up skipping over some smaller number of bytes, possibly <code>0</code>.
+ * This may result from any of a number of conditions; reaching end of file
+ * before <code>n</code> bytes have been skipped is only one possibility.
+ * The actual number of bytes skipped is returned. If {@code n} is
+ * negative, the {@code skip} method for class {@code InputStream} always
+ * returns 0, and no bytes are skipped. Subclasses may handle the negative
+ * value differently.
+ * <p>
+ * <p> The <code>skip</code> method of this class creates a
+ * byte array and then repeatedly reads into it until <code>n</code> bytes
+ * have been read or the end of the stream has been reached. Subclasses are
+ * encouraged to provide a more efficient implementation of this method.
+ * For instance, the implementation may depend on the ability to seek.
+ *
+ * @param n the number of bytes to be skipped.
+ * @return the actual number of bytes skipped.
+ * @throws IOException if the stream does not support seek,
+ * or if some other I/O error occurs.
+ */
+ @Override
+ public long skip(long n) throws IOException {
+
+ return in.skip(n);
+ }
+
+ /**
+ * Returns an estimate of the number of bytes that can be read (or
+ * skipped over) from this input stream without blocking by the next
+ * invocation of a method for this input stream. The next invocation
+ * might be the same thread or another thread. A single read or skip of this
+ * many bytes will not block, but may read or skip fewer bytes.
+ * <p>
+ * <p> Note that while some implementations of {@code InputStream} will return
+ * the total number of bytes in the stream, many will not. It is
+ * never correct to use the return value of this method to allocate
+ * a buffer intended to hold all data in this stream.
+ * <p>
+ * <p> A subclass' implementation of this method may choose to throw an
+ * {@link IOException} if this input stream has been closed by
+ * invoking the {@link #close()} method.
+ * <p>
+ * <p> The {@code available} method for class {@code InputStream} always
+ * returns {@code 0}.
+ * <p>
+ * <p> This method should be overridden by subclasses.
+ *
+ * @return an estimate of the number of bytes that can be read (or skipped
+ * over) from this input stream without blocking or {@code 0} when
+ * it reaches the end of the input stream.
+ * @throws IOException if an I/O error occurs.
+ */
+ @Override
+ public int available() throws IOException {
+
+ return in.available();
+ }
+
+ /**
+ * Closes this input stream and releases any system resources associated
+ * with the stream.
+ * <p>
+ * <p> The <code>close</code> method of <code>InputStream</code> does
+ * nothing.
+ *
+ * @throws IOException if an I/O error occurs.
+ */
+ @Override
+ public void close() throws IOException {
+ in.close();
+ }
+
+ /**
+ * Marks the current position in this input stream. A subsequent call to
+ * the <code>reset</code> method repositions this stream at the last marked
+ * position so that subsequent reads re-read the same bytes.
+ * <p>
+ * <p> The <code>readlimit</code> arguments tells this input stream to
+ * allow that many bytes to be read before the mark position gets
+ * invalidated.
+ * <p>
+ * <p> The general contract of <code>mark</code> is that, if the method
+ * <code>markSupported</code> returns <code>true</code>, the stream somehow
+ * remembers all the bytes read after the call to <code>mark</code> and
+ * stands ready to supply those same bytes again if and whenever the method
+ * <code>reset</code> is called. However, the stream is not required to
+ * remember any data at all if more than <code>readlimit</code> bytes are
+ * read from the stream before <code>reset</code> is called.
+ * <p>
+ * <p> Marking a closed stream should not have any effect on the stream.
+ * <p>
+ * <p> The <code>mark</code> method of <code>InputStream</code> does
+ * nothing.
+ *
+ * @param readlimit the maximum limit of bytes that can be read before
+ * the mark position becomes invalid.
+ * @see InputStream#reset()
+ */
+ @Override
+ public synchronized void mark(int readlimit) {
+ in.mark(readlimit);
+ }
+
+ /**
+ * Repositions this stream to the position at the time the
+ * <code>mark</code> method was last called on this input stream.
+ * <p>
+ * <p> The general contract of <code>reset</code> is:
+ * <p>
+ * <ul>
+ * <li> If the method <code>markSupported</code> returns
+ * <code>true</code>, then:
+ * <p>
+ * <ul><li> If the method <code>mark</code> has not been called since
+ * the stream was created, or the number of bytes read from the stream
+ * since <code>mark</code> was last called is larger than the argument
+ * to <code>mark</code> at that last call, then an
+ * <code>IOException</code> might be thrown.
+ * <p>
+ * <li> If such an <code>IOException</code> is not thrown, then the
+ * stream is reset to a state such that all the bytes read since the
+ * most recent call to <code>mark</code> (or since the start of the
+ * file, if <code>mark</code> has not been called) will be resupplied
+ * to subsequent callers of the <code>read</code> method, followed by
+ * any bytes that otherwise would have been the next input data as of
+ * the time of the call to <code>reset</code>. </ul>
+ * <p>
+ * <li> If the method <code>markSupported</code> returns
+ * <code>false</code>, then:
+ * <p>
+ * <ul><li> The call to <code>reset</code> may throw an
+ * <code>IOException</code>.
+ * <p>
+ * <li> If an <code>IOException</code> is not thrown, then the stream
+ * is reset to a fixed state that depends on the particular type of the
+ * input stream and how it was created. The bytes that will be supplied
+ * to subsequent callers of the <code>read</code> method depend on the
+ * particular type of the input stream. </ul></ul>
+ * <p>
+ * <p>The method <code>reset</code> for class <code>InputStream</code>
+ * does nothing except throw an <code>IOException</code>.
+ *
+ * @throws IOException if this stream has not been marked or if the
+ * mark has been invalidated.
+ * @see InputStream#mark(int)
+ * @see IOException
+ */
+ @Override
+ public synchronized void reset() throws IOException {
+ in.reset();
+ }
+
+ /**
+ * Tests if this input stream supports the <code>mark</code> and
+ * <code>reset</code> methods. Whether or not <code>mark</code> and
+ * <code>reset</code> are supported is an invariant property of a
+ * particular input stream instance. The <code>markSupported</code> method
+ * of <code>InputStream</code> returns <code>false</code>.
+ *
+ * @return <code>true</code> if this stream instance supports the mark
+ * and reset methods; <code>false</code> otherwise.
+ * @see InputStream#mark(int)
+ * @see InputStream#reset()
+ */
+ @Override
+ public boolean markSupported() {
+ return in.markSupported();
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShell.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShell.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShell.java
new file mode 100644
index 0000000..44ad28c
--- /dev/null
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShell.java
@@ -0,0 +1,427 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell.cli;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterables;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.PosixParser;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.log4j.PropertyConfigurator;
+import org.apache.metron.stellar.common.shell.DefaultStellarAutoCompleter;
+import org.apache.metron.stellar.common.shell.DefaultStellarShellExecutor;
+import org.apache.metron.stellar.common.shell.StellarAutoCompleter;
+import org.apache.metron.stellar.common.shell.StellarResult;
+import org.apache.metron.stellar.common.shell.StellarShellExecutor;
+import org.apache.metron.stellar.common.utils.JSONUtils;
+import org.jboss.aesh.complete.CompleteOperation;
+import org.jboss.aesh.complete.Completion;
+import org.jboss.aesh.console.AeshConsoleCallback;
+import org.jboss.aesh.console.Console;
+import org.jboss.aesh.console.ConsoleOperation;
+import org.jboss.aesh.console.Prompt;
+import org.jboss.aesh.console.settings.SettingsBuilder;
+import org.jboss.aesh.terminal.CharacterType;
+import org.jboss.aesh.terminal.Color;
+import org.jboss.aesh.terminal.TerminalCharacter;
+import org.jboss.aesh.terminal.TerminalColor;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import static org.apache.metron.stellar.dsl.Context.Capabilities.CONSOLE;
+import static org.apache.metron.stellar.dsl.Context.Capabilities.GLOBAL_CONFIG;
+import static org.apache.metron.stellar.dsl.Context.Capabilities.SHELL_VARIABLES;
+
+/**
+ * A REPL environment for Stellar.
+ *
+ * Useful for debugging Stellar expressions.
+ */
+public class StellarShell extends AeshConsoleCallback implements Completion {
+
+ static final String WELCOME = "Stellar, Go!\n" +
+ "Functions are loading lazily in the background and will be unavailable until loaded fully.";
+
+ private List<TerminalCharacter> EXPRESSION_PROMPT = new ArrayList<TerminalCharacter>()
+ {{
+ add(new TerminalCharacter('[', new TerminalColor(Color.RED, Color.DEFAULT)));
+ add(new TerminalCharacter('S', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
+ add(new TerminalCharacter('t', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
+ add(new TerminalCharacter('e', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
+ add(new TerminalCharacter('l', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
+ add(new TerminalCharacter('l', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
+ add(new TerminalCharacter('a', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
+ add(new TerminalCharacter('r', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.BOLD));
+ add(new TerminalCharacter(']', new TerminalColor(Color.RED, Color.DEFAULT)));
+ add(new TerminalCharacter('>', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.UNDERLINE));
+ add(new TerminalCharacter('>', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.UNDERLINE));
+ add(new TerminalCharacter('>', new TerminalColor(Color.GREEN, Color.DEFAULT), CharacterType.UNDERLINE));
+ add(new TerminalCharacter(' ', new TerminalColor(Color.DEFAULT, Color.DEFAULT)));
+ }};
+
+ public static final String ERROR_PROMPT = "[!] ";
+ public static final String STELLAR_PROPERTIES_FILENAME = "stellar.properties";
+
+ /**
+ * Executes Stellar expressions for the shell environment.
+ */
+ private StellarShellExecutor executor;
+
+ /**
+ * The Aesh shell console.
+ */
+ private Console console;
+
+ /**
+ * Provides auto-complete functionality.
+ */
+ private StellarAutoCompleter autoCompleter;
+
+ /**
+ * Execute the Stellar REPL.
+ */
+ public static void main(String[] args) throws Exception {
+ StellarShell shell = new StellarShell(args);
+ shell.run();
+ }
+
+ /**
+ * Create a Stellar REPL.
+ * @param args The commmand-line arguments.
+ */
+ public StellarShell(String[] args) throws Exception {
+
+ // define valid command-line options
+ CommandLineParser parser = new PosixParser();
+ Options options = defineCommandLineOptions();
+ CommandLine commandLine = parser.parse(options, args);
+
+ // print help
+ if(commandLine.hasOption("h")) {
+ HelpFormatter formatter = new HelpFormatter();
+ formatter.printHelp("stellar", options);
+ System.exit(0);
+ }
+
+ // validate the command line options
+ try {
+ StellarShellOptionsValidator.validateOptions(commandLine);
+
+ } catch(IllegalArgumentException e){
+ System.err.println(e.getMessage());
+ System.exit(1);
+ }
+
+ // setup logging, if specified
+ if(commandLine.hasOption("l")) {
+ PropertyConfigurator.configure(commandLine.getOptionValue("l"));
+ }
+
+ console = createConsole(commandLine);
+ autoCompleter = new DefaultStellarAutoCompleter();
+ Properties props = getStellarProperties(commandLine);
+ executor = createExecutor(commandLine, console, props, autoCompleter);
+ loadVariables(commandLine, executor);
+ console.setPrompt(new Prompt(EXPRESSION_PROMPT));
+ console.addCompletion(this);
+ console.setConsoleCallback(this);
+ }
+
+ /**
+ * @return The valid command line options.
+ */
+ private Options defineCommandLineOptions() {
+ Options options = new Options();
+ options.addOption(
+ "z",
+ "zookeeper",
+ true,
+ "Zookeeper URL fragment in the form [HOSTNAME|IPADDRESS]:PORT");
+ options.addOption(
+ "v",
+ "variables",
+ true,
+ "File containing a JSON Map of variables");
+ options.addOption(
+ "irc",
+ "inputrc",
+ true,
+ "File containing the inputrc if not the default ~/.inputrc");
+ options.addOption(
+ "na",
+ "no_ansi",
+ false,
+ "Make the input prompt not use ANSI colors.");
+ options.addOption(
+ "h",
+ "help",
+ false,
+ "Print help");
+ options.addOption(
+ "p",
+ "properties",
+ true,
+ "File containing Stellar properties");
+ Option log4j = new Option(
+ "l",
+ "log4j",
+ true,
+ "The log4j properties file to load");
+ log4j.setArgName("FILE");
+ log4j.setRequired(false);
+ options.addOption(log4j);
+
+ return options;
+ }
+
+ /**
+ * Loads any variables defined in an external file.
+ * @param commandLine The command line arguments.
+ * @param executor The stellar executor.
+ * @throws IOException
+ */
+ private static void loadVariables(
+ CommandLine commandLine,
+ StellarShellExecutor executor) throws IOException {
+
+ if(commandLine.hasOption("v")) {
+
+ // load variables defined in a file
+ String variablePath = commandLine.getOptionValue("v");
+ Map<String, Object> variables = JSONUtils.INSTANCE.load(
+ new File(variablePath),
+ new TypeReference<Map<String, Object>>() {});
+
+ // for each variable...
+ for(Map.Entry<String, Object> kv : variables.entrySet()) {
+ String variable = kv.getKey();
+ Object value = kv.getValue();
+
+ // define the variable - no expression available
+ executor.assign(variable, value, Optional.empty());
+ }
+ }
+ }
+
+ /**
+ * Creates the Stellar execution environment.
+ * @param commandLine The command line arguments.
+ * @param console The console which drives the REPL.
+ * @param properties Stellar properties.
+ */
+ private StellarShellExecutor createExecutor(
+ CommandLine commandLine,
+ Console console,
+ Properties properties,
+ StellarAutoCompleter autoCompleter) throws Exception {
+
+ // setup zookeeper client
+ Optional<String> zookeeperUrl = Optional.empty();
+ if(commandLine.hasOption("z")) {
+ zookeeperUrl = Optional.of(commandLine.getOptionValue("z"));
+ }
+
+ StellarShellExecutor executor = new DefaultStellarShellExecutor(properties, zookeeperUrl);
+
+ // the 'CONSOLE' capability is only available with the CLI REPL
+ executor.getContext().addCapability(CONSOLE, () -> console);
+
+ // allows some Stellar functions to access Stellar internals; should probably use %magics instead
+ executor.getContext().addCapability(SHELL_VARIABLES, () -> executor.getState());
+
+ // register the auto-completer to be notified when needed
+ executor.addSpecialListener( (special) -> autoCompleter.addCandidateFunction(special.getCommand()));
+ executor.addFunctionListener( (function) -> autoCompleter.addCandidateFunction(function.getName()));
+ executor.addVariableListener((name, val) -> autoCompleter.addCandidateVariable(name));
+
+ executor.init();
+ return executor;
+ }
+
+ /**
+ * Creates the REPL's console.
+ * @param commandLine The command line options.
+ */
+ private Console createConsole(CommandLine commandLine) {
+
+ // console settings
+ boolean useAnsi = !commandLine.hasOption("na");
+ SettingsBuilder settings = new SettingsBuilder().enableAlias(true)
+ .enableMan(true)
+ .ansi(useAnsi)
+ .parseOperators(false)
+ .inputStream(PausableInput.INSTANCE);
+
+ if(commandLine.hasOption("irc")) {
+ settings = settings.inputrc(new File(commandLine.getOptionValue("irc")));
+ }
+
+ return new Console(settings.create());
+ }
+
+ /**
+ * Retrieves the Stellar properties. The properties are either loaded from a file in
+ * the classpath or a set of defaults are used.
+ */
+ private Properties getStellarProperties(CommandLine commandLine) throws IOException {
+ Properties properties = new Properties();
+
+ if (commandLine.hasOption("p")) {
+ // attempt to load properties from a file specified on the command-line
+ try (InputStream in = new FileInputStream(commandLine.getOptionValue("p"))) {
+ if(in != null) {
+ properties.load(in);
+ }
+ }
+
+ } else {
+ // otherwise attempt to load properties from the classpath
+ try (InputStream in = getClass().getClassLoader().getResourceAsStream(STELLAR_PROPERTIES_FILENAME)) {
+ if(in != null) {
+ properties.load(in);
+ }
+ }
+ }
+
+ return properties;
+ }
+
+ /**
+ * Handles the main loop for the REPL.
+ */
+ public void run() {
+ // welcome message
+ writeLine(WELCOME);
+
+ // print the globals if we got 'em
+ executor.getContext()
+ .getCapability(GLOBAL_CONFIG, false)
+ .ifPresent(conf -> writeLine(conf.toString()));
+
+ console.start();
+ }
+
+ /**
+ * Quits the console.
+ */
+ private void handleQuit() {
+ try {
+ console.stop();
+ } catch (Throwable e) {
+ e.printStackTrace();
+ }
+ }
+
+ private void writeLine(String out) {
+ console.getShell().out().println(out);
+ }
+
+ @Override
+ public int execute(ConsoleOperation output) throws InterruptedException {
+
+ // grab the user the input
+ String expression = StringUtils.trimToEmpty(output.getBuffer());
+ if(StringUtils.isNotBlank(expression) ) {
+
+ // execute the expression
+ StellarResult result = executor.execute(expression);
+
+ if(result.isSuccess()) {
+ // on success
+ result.getValue().ifPresent(v -> writeLine(v.toString()));
+
+ } else if (result.isError()) {
+ // on error
+ result.getException().ifPresent(e -> writeLine(ERROR_PROMPT + e.getMessage()));
+ result.getException().ifPresent(e -> e.printStackTrace());
+
+ } else if(result.isTerminate()) {
+ // on quit
+ handleQuit();
+
+ } else {
+ // should never happen
+ throw new IllegalStateException("An execution result is neither a success nor a failure. Please file a bug report.");
+ }
+ }
+
+ return 0;
+ }
+
+ /**
+ * Performs auto-completion for the shell.
+ * @param completeOperation The auto-complete operation.
+ */
+ @Override
+ public void complete(CompleteOperation completeOperation) {
+ String buffer = completeOperation.getBuffer();
+ final String lastToken = getLastToken(buffer);
+ Iterable<String> candidates = autoCompleter.autoComplete(buffer);
+
+ // transform the candidates into valid completions
+ if(candidates != null && !Iterables.isEmpty(candidates)) {
+ for(String candidate : candidates) {
+ String completion = stripOff(buffer, lastToken) + candidate;
+ completeOperation.addCompletionCandidate(completion);
+ }
+ }
+ }
+
+ private static String getLastToken(String buffer) {
+ String lastToken = Iterables.getLast(Splitter.on(" ").split(buffer), null);
+ return lastToken.trim();
+ }
+
+ private static String stripOff(String baseString, String lastBit) {
+ int index = baseString.lastIndexOf(lastBit);
+ if(index < 0) {
+ return baseString;
+ }
+ return baseString.substring(0, index);
+ }
+
+ /**
+ * @return The executor of Stellar expressions.
+ */
+ public StellarShellExecutor getExecutor() {
+ return executor;
+ }
+
+ /**
+ * @return The console.
+ */
+ public Console getConsole() {
+ return console;
+ }
+}
http://git-wip-us.apache.org/repos/asf/metron/blob/2dd01b17/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShellOptionsValidator.java
----------------------------------------------------------------------
diff --git a/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShellOptionsValidator.java b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShellOptionsValidator.java
new file mode 100644
index 0000000..585add5
--- /dev/null
+++ b/metron-stellar/stellar-common/src/main/java/org/apache/metron/stellar/common/shell/cli/StellarShellOptionsValidator.java
@@ -0,0 +1,127 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF 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 org.apache.metron.stellar.common.shell.cli;
+
+import java.io.File;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.function.Predicate;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import com.google.common.base.Splitter;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.validator.routines.InetAddressValidator;
+
+public class StellarShellOptionsValidator {
+
+ private static final Pattern validPortPattern = Pattern.compile("(^.*)[:](\\d+)$");
+ private static final Predicate<String> hostnameValidator = hostname -> {
+ if(StringUtils.isEmpty(hostname)) {
+ return false;
+ }
+ try {
+ InetAddress add = InetAddress.getByName(hostname);
+ return true;
+ } catch (UnknownHostException e) {
+ return false;
+ }
+ };
+
+
+
+ private static final InetAddressValidator inetAddressValidator = InetAddressValidator
+ .getInstance();
+
+ /**
+ * Validates Stellar CLI Options.
+ */
+ public static void validateOptions(CommandLine commandLine) throws IllegalArgumentException {
+ if (commandLine.hasOption('z')) {
+ validateZookeeperOption(commandLine.getOptionValue('z'));
+ }
+ // v, irc, p are files
+ if (commandLine.hasOption('v')) {
+ validateFileOption("v", commandLine.getOptionValue('v'));
+ }
+ if (commandLine.hasOption("irc")) {
+ validateFileOption("irc", commandLine.getOptionValue("irc"));
+ }
+ if (commandLine.hasOption('p')) {
+ validateFileOption("p", commandLine.getOptionValue('p'));
+ }
+
+ }
+
+ /**
+ * Zookeeper argument should be in the form [HOST|IP]:PORT.
+ *
+ * @param zMulti the zookeeper url fragment
+ */
+ private static void validateZookeeperOption(String zMulti) throws IllegalArgumentException {
+ for(String z : Splitter.on(",").split(zMulti)) {
+ Matcher matcher = validPortPattern.matcher(z);
+ boolean hasPort = z.contains(":");
+ if (hasPort && !matcher.matches()) {
+ throw new IllegalArgumentException(String.format("Zookeeper option must have valid port: %s", z));
+ }
+
+ if (hasPort && matcher.groupCount() != 2) {
+ throw new IllegalArgumentException(
+ String.format("Zookeeper Option must be in the form of [HOST|IP]:PORT %s", z));
+ }
+ String name = hasPort?matcher.group(1):z;
+ Integer port = hasPort?Integer.parseInt(matcher.group(2)):null;
+
+ if (!hostnameValidator.test(name) && !inetAddressValidator.isValid(name)) {
+ throw new IllegalArgumentException(
+ String.format("Zookeeper Option %s is not a valid host name or ip address %s", name, z));
+ }
+
+ if (hasPort && (port == 0 || port > 65535)) {
+ throw new IllegalArgumentException(
+ String.format("Zookeeper Option %s port is not valid", z));
+ }
+ }
+ }
+
+ /**
+ * File options must exist and be readable.
+ *
+ * @param option name of the option
+ * @param fileName the file name
+ */
+ private static void validateFileOption(String option, String fileName)
+ throws IllegalArgumentException {
+ File file = new File(fileName);
+ if (!file.exists()) {
+ throw new IllegalArgumentException(
+ String.format("%s: File %s doesn't exist", option, fileName));
+ }
+ if (!file.canRead()) {
+ throw new IllegalArgumentException(
+ String.format("%s: File %s is not readable", option, fileName));
+ }
+ }
+}
+
+