samples/apps/src/main/java/org/apache/edgent/samples/apps/ApplicationUtilities.java - incubator-retired-edgent - Git at Google

 /*
 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.edgent.samples.apps;

 import java.io.File;
 import java.util.Date;
 import java.util.HashSet;
 import java.util.Properties;
 import java.util.Set;

 import org.apache.edgent.analytics.sensors.Range;
 import org.apache.edgent.analytics.sensors.Ranges;
 import org.apache.edgent.connectors.file.FileStreams;
 import org.apache.edgent.connectors.file.FileWriterCycleConfig;
 import org.apache.edgent.connectors.file.FileWriterFlushConfig;
 import org.apache.edgent.connectors.file.FileWriterPolicy;
 import org.apache.edgent.connectors.file.FileWriterRetentionConfig;
 import org.apache.edgent.function.Predicate;
 import org.apache.edgent.function.Supplier;
 import org.apache.edgent.topology.TStream;

 /**
  * Some general purpose application configuration driven utilities.
  * <p>
  * Utilities include:
  * <ul>
  * <li>Get a property name for a sensor configuration item</li>
  * <li>Get a Range value for a sensor range item</li>
  * <li>Log a stream</li>
  * <li>Conditionally trace a stream</li>
  * </ul>
  */
 public class ApplicationUtilities {

     private final Properties props;

     public ApplicationUtilities(Properties props) {
         this.props = props;
     }

     private Properties config() {
         return props;
     }

     /**
      * Trace a stream to System.out if the sensor id's "label" has been configured
      * to enable tracing.
      * <p>
      * If tracing has not been enabled in the config, the topology will not
      * be augmented to trace the stream.
      *
      * @param <T> Tuple type
      * @param stream the stream to trace
      * @param sensorId the sensor id
      * @param label some unique label
      * @return the input stream
      */
     public <T> TStream<T> traceStream(TStream<T> stream, String sensorId, Supplier<String> label) {
         return traceStream(stream, () -> sensorId+"."+label.get());
     }

     /**
      * Trace a stream to System.out if the "label" has been configured
      * to enable tracing.
      * <p>
      * If tracing has not been enabled in the config, the topology will not
      * be augmented to trace the stream.
      *
      * @param <T> Tuple type
      * @param stream the stream to trace
      * @param label some unique label
      * @return the input stream
      */
     public <T> TStream<T> traceStream(TStream<T> stream, Supplier<String> label) {
         if (includeTraceStreamOps(label.get())) {
             TStream<?> s = stream.filter(traceTuplesFn(label.get())).tag(label.get()+".trace");
             s.peek(sample -> System.out.println(String.format("%s: %s", label.get(), sample.toString())));
         }
         return stream;
     }

     private boolean includeTraceStreamOps(String label) {
         String includesCsv = config().getProperty("stream.tracing.includes.csv", "");
         String includesRegex = config().getProperty("stream.tracing.includes.regex", "");
         String excludesCsv = config().getProperty("stream.tracing.excludes.csv", "");
         String excludesRegex = config().getProperty("stream.tracing.excludes.regex", "");

         Set<String> includesSet = new HashSet<>();
         for (String s : includesCsv.split(","))
             includesSet.add(s.trim());
         Set<String> excludesSet = new HashSet<>();
         for (String s : excludesCsv.split(","))
             excludesSet.add(s.trim());

         boolean isIncluded = false;
         if (includesSet.contains(label) || label.matches(includesRegex))
             isIncluded = true;
         if (excludesSet.contains(label) || label.matches(excludesRegex))
             isIncluded = false;

         return isIncluded;
     }

     private <T> Predicate<T> traceTuplesFn(String label) {
         return tuple -> true; // TODO make dynamic config; affected by "label" value
         // check label for match against csv or regex from props
     }

     /**
      * Get the property name for a sensor's configuration item.
      * @param sensorId the sensor's id
      * @param label the label for an instance of "kind" (e.g., "tempThreshold")
      * @param kind the kind of configuration item (e.g., "range")
      * @return the configuration property name
      */
     public String getSensorPropertyName(String sensorId, String label, String kind) {
         String name = kind + "." + label;  // kind.label
         if (sensorId!=null && !sensorId.isEmpty())
             name = sensorId + "." + name;  // sensorId.kind.label
         return name;
     }

     private String getSensorConfigValue(String sensorId, String label, String kind) {
         String name = getSensorPropertyName(sensorId, label, kind);
         String val = config().getProperty(name);
         if (val==null)
             throw new IllegalArgumentException("Missing configuration property "+name);
         return val;
     }

     /**
      * Get the Range for a sensor range configuration item.
      * @param sensorId the sensor's id
      * @param label the range's label
      * @return the Range
      */
     public Range<Integer> getRangeInteger(String sensorId, String label) {
         String val = getSensorConfigValue(sensorId, label, "range");
         return Ranges.valueOfInteger(val);
     }

     /**
      * Get the Range for a sensor range configuration item.
      * @param sensorId the sensor's id
      * @param label the range's label
      * @return the Range
      */
     public Range<Byte> getRangeByte(String sensorId, String label) {
         String val = getSensorConfigValue(sensorId, label, "range");
         return Ranges.valueOfByte(val);
     }

     /**
      * Get the Range for a sensor range configuration item.
      * @param sensorId the sensor's id
      * @param label the range's label
      * @return the Range
      */
     public Range<Short> getRangeShort(String sensorId, String label) {
         String val = getSensorConfigValue(sensorId, label, "range");
         return Ranges.valueOfShort(val);
     }

     /**
      * Get the Range for a sensor range configuration item.
      * @param sensorId the sensor's id
      * @param label the range's label
      * @return the Range
      */
     public Range<Float> getRangeFloat(String sensorId, String label) {
         String val = getSensorConfigValue(sensorId, label, "range");
         return Ranges.valueOfFloat(val);
     }

     /**
      * Get the Range for a sensor range configuration item.
      * @param sensorId the sensor's id
      * @param label the range's label
      * @return the Range
      */
     public Range<Double> getRangeDouble(String sensorId, String label) {
         String val = getSensorConfigValue(sensorId, label, "range");
         return Ranges.valueOfDouble(val);
     }

     /**
      * Log every tuple on the stream using the {@code FileStreams} connector.
      * <p>
      * The logs are added to the directory as specified
      * by the "application.log.dir" property.
      * The directory will be created as needed.
      * <p>
      * The "active" (open / being written) log file name is {@code .<baseName>}.
      * <br>
      * Completed stable logs have a name of {@code <baseName>_YYYYMMDD_HHMMSS}.
      * <p>
      * The log entry format being used is:
      * {@code [<date>] [<eventTag>] <tuple>.toString()}
      * <p>
      * See {@link FileStreams#textFileWriter(TStream, org.apache.edgent.function.Supplier, org.apache.edgent.function.Supplier)}
      *
      * @param <T> Tuple type
      * @param stream the TStream
      * @param baseName the base log name
      * @param eventTag a tag that gets added to the log entry
      * @return the input stream
      */
     public <T> TStream<T> logStream(TStream<T> stream, String eventTag, String baseName) {
         // Define the writer policy.
         // TODO could make the policy configurable via config()
         FileWriterPolicy<String> policy = new FileWriterPolicy<String>(
                 FileWriterFlushConfig.newTimeBasedConfig(2_000/*msec*/), // flush every 2sec
                 FileWriterCycleConfig.newFileSizeBasedConfig(10_000),  // new file every 10KB
                 FileWriterRetentionConfig.newFileCountBasedConfig(1)   // retain 1 file
                 );

         // Compose the base file pathname
         File dir = new File(config().getProperty("application.log.dir"));
         String basePathname = new File(dir, baseName).toString();

         // Transform the stream to a TStream<String> of string log entry values
         TStream<String> stringEntries = stream.map(sample -> String.format("[%s] [%s] %s", new Date().toString(), eventTag, sample.toString()))
                 .tag(baseName+".log");

         // Use the FileStreams connector to write the logs.
         //
         // A hack for getting the log directories created at runtime
         // TODO add another policy thing... or simply make textFileWriter do it?
         //
         FileStreams.textFileWriter(stringEntries,
                 () -> { if (!dir.exists()) dir.mkdirs();
                         return basePathname;
                       },
                 () -> policy);

         return stream;
     }

 }
	/*
	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.edgent.samples.apps;

	import java.io.File;
	import java.util.Date;
	import java.util.HashSet;
	import java.util.Properties;
	import java.util.Set;

	import org.apache.edgent.analytics.sensors.Range;
	import org.apache.edgent.analytics.sensors.Ranges;
	import org.apache.edgent.connectors.file.FileStreams;
	import org.apache.edgent.connectors.file.FileWriterCycleConfig;
	import org.apache.edgent.connectors.file.FileWriterFlushConfig;
	import org.apache.edgent.connectors.file.FileWriterPolicy;
	import org.apache.edgent.connectors.file.FileWriterRetentionConfig;
	import org.apache.edgent.function.Predicate;
	import org.apache.edgent.function.Supplier;
	import org.apache.edgent.topology.TStream;

	/**
	* Some general purpose application configuration driven utilities.
	* <p>
	* Utilities include:
	* <ul>
	* <li>Get a property name for a sensor configuration item</li>
	* <li>Get a Range value for a sensor range item</li>
	* <li>Log a stream</li>
	* <li>Conditionally trace a stream</li>
	* </ul>
	*/
	public class ApplicationUtilities {

	private final Properties props;

	public ApplicationUtilities(Properties props) {
	this.props = props;
	}

	private Properties config() {
	return props;
	}

	/**
	* Trace a stream to System.out if the sensor id's "label" has been configured
	* to enable tracing.
	* <p>
	* If tracing has not been enabled in the config, the topology will not
	* be augmented to trace the stream.
	*
	* @param <T> Tuple type
	* @param stream the stream to trace
	* @param sensorId the sensor id
	* @param label some unique label
	* @return the input stream
	*/
	public <T> TStream<T> traceStream(TStream<T> stream, String sensorId, Supplier<String> label) {
	return traceStream(stream, () -> sensorId+"."+label.get());
	}

	/**
	* Trace a stream to System.out if the "label" has been configured
	* to enable tracing.
	* <p>
	* If tracing has not been enabled in the config, the topology will not
	* be augmented to trace the stream.
	*
	* @param <T> Tuple type
	* @param stream the stream to trace
	* @param label some unique label
	* @return the input stream
	*/
	public <T> TStream<T> traceStream(TStream<T> stream, Supplier<String> label) {
	if (includeTraceStreamOps(label.get())) {
	TStream<?> s = stream.filter(traceTuplesFn(label.get())).tag(label.get()+".trace");
	s.peek(sample -> System.out.println(String.format("%s: %s", label.get(), sample.toString())));
	}
	return stream;
	}

	private boolean includeTraceStreamOps(String label) {
	String includesCsv = config().getProperty("stream.tracing.includes.csv", "");
	String includesRegex = config().getProperty("stream.tracing.includes.regex", "");
	String excludesCsv = config().getProperty("stream.tracing.excludes.csv", "");
	String excludesRegex = config().getProperty("stream.tracing.excludes.regex", "");

	Set<String> includesSet = new HashSet<>();
	for (String s : includesCsv.split(","))
	includesSet.add(s.trim());
	Set<String> excludesSet = new HashSet<>();
	for (String s : excludesCsv.split(","))
	excludesSet.add(s.trim());

	boolean isIncluded = false;
	if (includesSet.contains(label) \|\| label.matches(includesRegex))
	isIncluded = true;
	if (excludesSet.contains(label) \|\| label.matches(excludesRegex))
	isIncluded = false;

	return isIncluded;
	}

	private <T> Predicate<T> traceTuplesFn(String label) {
	return tuple -> true; // TODO make dynamic config; affected by "label" value
	// check label for match against csv or regex from props
	}

	/**
	* Get the property name for a sensor's configuration item.
	* @param sensorId the sensor's id
	* @param label the label for an instance of "kind" (e.g., "tempThreshold")
	* @param kind the kind of configuration item (e.g., "range")
	* @return the configuration property name
	*/
	public String getSensorPropertyName(String sensorId, String label, String kind) {
	String name = kind + "." + label; // kind.label
	if (sensorId!=null && !sensorId.isEmpty())
	name = sensorId + "." + name; // sensorId.kind.label
	return name;
	}

	private String getSensorConfigValue(String sensorId, String label, String kind) {
	String name = getSensorPropertyName(sensorId, label, kind);
	String val = config().getProperty(name);
	if (val==null)
	throw new IllegalArgumentException("Missing configuration property "+name);
	return val;
	}

	/**
	* Get the Range for a sensor range configuration item.
	* @param sensorId the sensor's id
	* @param label the range's label
	* @return the Range
	*/
	public Range<Integer> getRangeInteger(String sensorId, String label) {
	String val = getSensorConfigValue(sensorId, label, "range");
	return Ranges.valueOfInteger(val);
	}

	/**
	* Get the Range for a sensor range configuration item.
	* @param sensorId the sensor's id
	* @param label the range's label
	* @return the Range
	*/
	public Range<Byte> getRangeByte(String sensorId, String label) {
	String val = getSensorConfigValue(sensorId, label, "range");
	return Ranges.valueOfByte(val);
	}

	/**
	* Get the Range for a sensor range configuration item.
	* @param sensorId the sensor's id
	* @param label the range's label
	* @return the Range
	*/
	public Range<Short> getRangeShort(String sensorId, String label) {
	String val = getSensorConfigValue(sensorId, label, "range");
	return Ranges.valueOfShort(val);
	}

	/**
	* Get the Range for a sensor range configuration item.
	* @param sensorId the sensor's id
	* @param label the range's label
	* @return the Range
	*/
	public Range<Float> getRangeFloat(String sensorId, String label) {
	String val = getSensorConfigValue(sensorId, label, "range");
	return Ranges.valueOfFloat(val);
	}

	/**
	* Get the Range for a sensor range configuration item.
	* @param sensorId the sensor's id
	* @param label the range's label
	* @return the Range
	*/
	public Range<Double> getRangeDouble(String sensorId, String label) {
	String val = getSensorConfigValue(sensorId, label, "range");
	return Ranges.valueOfDouble(val);
	}

	/**
	* Log every tuple on the stream using the {@code FileStreams} connector.
	* <p>
	* The logs are added to the directory as specified
	* by the "application.log.dir" property.
	* The directory will be created as needed.
	* <p>
	* The "active" (open / being written) log file name is {@code .<baseName>}.
	* <br>
	* Completed stable logs have a name of {@code <baseName>_YYYYMMDD_HHMMSS}.
	* <p>
	* The log entry format being used is:
	* {@code [<date>] [<eventTag>] <tuple>.toString()}
	* <p>
	* See {@link FileStreams#textFileWriter(TStream, org.apache.edgent.function.Supplier, org.apache.edgent.function.Supplier)}
	*
	* @param <T> Tuple type
	* @param stream the TStream
	* @param baseName the base log name
	* @param eventTag a tag that gets added to the log entry
	* @return the input stream
	*/
	public <T> TStream<T> logStream(TStream<T> stream, String eventTag, String baseName) {
	// Define the writer policy.
	// TODO could make the policy configurable via config()
	FileWriterPolicy<String> policy = new FileWriterPolicy<String>(
	FileWriterFlushConfig.newTimeBasedConfig(2_000/msec/), // flush every 2sec
	FileWriterCycleConfig.newFileSizeBasedConfig(10_000), // new file every 10KB
	FileWriterRetentionConfig.newFileCountBasedConfig(1) // retain 1 file
	);

	// Compose the base file pathname
	File dir = new File(config().getProperty("application.log.dir"));
	String basePathname = new File(dir, baseName).toString();

	// Transform the stream to a TStream<String> of string log entry values
	TStream<String> stringEntries = stream.map(sample -> String.format("[%s] [%s] %s", new Date().toString(), eventTag, sample.toString()))
	.tag(baseName+".log");

	// Use the FileStreams connector to write the logs.
	//
	// A hack for getting the log directories created at runtime
	// TODO add another policy thing... or simply make textFileWriter do it?
	//
	FileStreams.textFileWriter(stringEntries,
	() -> { if (!dir.exists()) dir.mkdirs();
	return basePathname;
	},
	() -> policy);

	return stream;
	}

	}