subprojects/s4-edsl/src/main/java/org/apache/s4/edsl/AppBuilder.java - incubator-retired-s4 - 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.s4.edsl;

 import java.lang.reflect.Array;
 import java.lang.reflect.Field;
 import java.util.Collection;
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;

 import org.apache.s4.base.Event;
 import org.apache.s4.core.App;
 import org.apache.s4.core.ProcessingElement;
 import org.apache.s4.core.Stream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.collect.LinkedListMultimap;
 import com.google.common.collect.Multimap;
 import com.google.common.collect.Sets;

 /**
  * Implementation of the S4 embedded domain-specific language (EDSL).
  *
  * <p>
  * To write an app extend this class and define the application graph using a chain of methods as follows:
  *
  * <pre>
  *    final public class MyApp extends BuilderS4DSL {
  *
  *     protected void onInit() {
  *
  *         pe("Consumer").type(ConsumerPE.class).asSingleton().
  *         pe("Producer").type(ProducerPE.class).timer().withPeriod(1, TimeUnit.MILLISECONDS).asSingleton().
  *         emit(SomeEvent.class).withKey("someKey").to("Consumer").
  *         build()
  *    }
  * </pre>
  *
  * <p>
  * A few things to notice:
  * <ul>
  * <li>Applications must extend class {@link BuilderS4DSL}
  * <li>The graph definition is implemented in the {@link App#onInit} method which is called by the container when the
  * application is loaded.
  * <li>PEs are defined using strings because they need to be referenced by other parts of the graph. By doing this, we
  * can create the whole application in a single chain of methods.
  * <li>To assign target streams to PE fields additional information may need to be provided using the {@code onField}
  * grammar token when there is an ambiguity. This will happen when a PE has more than one targetStream field with the
  * same {@link Event} type. Use the construct {@code emit(SomeEvent.class).onField("streamFieldName")}. If the PE
  * doesn't have a field named {@code "streamField"} whose stream parameter type is {@code someEvent)} then the parser
  * will fail to build the app.
  * <li>To configure a PE, set property values by chaining any number of {@code prop(name, value)} methods. The name
  * should match a PE field and the value will be parsed using the type of that field.
  * </ul>
  * <p>
  * Grammar:
  *
  * <pre>
  *  (pe , type , prop* , (fireOn , afterInterval? , afterNumEvents?)? , (timer, withPeriod)? ,
  *  (cache, size , expires? )? , asSingleton? , (emit, onField?,
  *  (withKey|withKeyFinder)?, to )*  )+ , build
  * </pre>
  *
  * <p>
  * See the <a href="http://code.google.com/p/diezel">Diezel</a> project for details.
  *
  */
 public class AppBuilder extends App {

     protected App app = this;

     static final Logger logger = LoggerFactory.getLogger(AppBuilder.class);

     private Multimap<ProcessingElement, StreamBuilder<? extends Event>> pe2stream = LinkedListMultimap.create();
     Set<StreamBuilder<? extends Event>> streamBuilders = Sets.newHashSet();

     /* Variables used to hold values from state to state. */
     ProcessingElement processingElement;
     String peName;
     Class<? extends Event> triggerEventType;
     long triggerInterval = 0;
     TimeUnit triggerTimeUnit;
     int cacheSize;
     StreamBuilder<? extends Event> streamBuilder;
     String propertyName, propertyValue;

     public static AppBuilder getAppBuilder() {
         return new BuilderS4DSL();
     }

     void addProperty(String name, String value) {
         propertyName = name;
         propertyValue = value;
         setField();
     }

     void addPe2Stream(ProcessingElement pe, StreamBuilder<? extends Event> st) {
         pe2stream.put(pe, st);
     }

     App buildApp() {

         /* Stream to PE writing. */
         for (StreamBuilder<? extends Event> sb : streamBuilders) {
             for (String peName : sb.pes) {
                 ProcessingElement pe = getPE(peName);
                 sb.stream.setPEs(pe);
             }
         }

         /* PE to Stream wiring. */
         Map<ProcessingElement, Collection<StreamBuilder<? extends Event>>> pe2streamMap = pe2stream.asMap();
         for (Map.Entry<ProcessingElement, Collection<StreamBuilder<? extends Event>>> entry : pe2streamMap.entrySet()) {
             ProcessingElement pe = entry.getKey();
             Collection<StreamBuilder<? extends Event>> streams = entry.getValue();

             if (pe != null && streams != null) {
                 try {
                     setStreamField(pe, streams);
                 } catch (Exception e) {
                     logger.error("Unable to build app.", e);
                     return null;
                 }
             }
         }

         return this;
     }

     /**
      * @param peName
      *            the peName to set
      */
     protected void setPeName(String peName) {
         this.peName = peName;
     }

     /*
      * Cannot create an abstract class in Diezel so for now, I just implement the abstract methods here. They need to be
      * overloaded by the app developer.
      */
     @Override
     protected void onStart() {
     }

     @Override
     protected void onInit() {
     }

     @Override
     protected void onClose() {
     }

     private <T extends ProcessingElement> void setField() {

         logger.debug("Adding property [{}] to PE of type [{}].", propertyName, processingElement.getClass().getName());

         Class<? extends ProcessingElement> type = processingElement.getClass();

         try {
             Field f = type.getDeclaredField(propertyName);
             f.setAccessible(true);
             logger.trace("Type: {}.", f.getType());
             logger.trace("GenericType: {}.", f.getGenericType());

             /* Set the field. */
             if (f.getType().getCanonicalName() == "long") {
                 f.setLong(processingElement, Long.parseLong(propertyValue));
                 return;
             } else if (f.getType().getCanonicalName() == "int") {
                 f.setInt(processingElement, Integer.parseInt(propertyValue));
                 return;
             } else if (f.getType().getCanonicalName() == "float") {
                 f.setFloat(processingElement, Float.parseFloat(propertyValue));
                 return;
             } else if (f.getType().getCanonicalName() == "double") {
                 f.setDouble(processingElement, Double.parseDouble(propertyValue));
                 return;
             } else if (f.getType().getCanonicalName() == "short") {
                 f.setShort(processingElement, Short.parseShort(propertyValue));
                 return;
             } else if (f.getType().getCanonicalName() == "byte") {
                 f.setByte(processingElement, Byte.parseByte(propertyValue));
                 return;
             } else if (f.getType().getCanonicalName() == "boolean") {
                 f.setBoolean(processingElement, Boolean.parseBoolean(propertyValue));
                 return;
             } else if (f.getType().getCanonicalName() == "char") {
                 f.setChar(processingElement, (char) Byte.parseByte(propertyValue));
                 return;
             } else if (f.getType().getCanonicalName() == "java.lang.String") {
                 f.set(processingElement, propertyValue);
                 return;
             }

             logger.error("Unable to set field named [{}] in PE of type [{}].", propertyName, type);
             throw new IllegalArgumentException();

             // production code should handle these exceptions more gracefully
         } catch (NoSuchFieldException e) {
             logger.error("There is no field named [{}] in PE of type [{}].", propertyName, type);
         } catch (Exception e) {
             logger.error("Couldn't set value for field [{}] in PE of type [{}].", propertyName, type);
         }
     }

     /* Set the stream fields in PE classes. Infer the field by checking the stream parameter type <? extends Event>. */
     private void setStreamField(ProcessingElement pe, Collection<StreamBuilder<? extends Event>> streams)
             throws Exception {

         /*
          * Create a map of the stream fields to the corresponding generic type. We will use this info to assign the
          * streams. If the field type matches the stream type and there is no ambiguity, then the assignment is easy. If
          * more than one field has the same type, then then we need to do more work.
          */
         Field[] fields = pe.getClass().getDeclaredFields();
         Multimap<String, Field> typeMap = LinkedListMultimap.create();
         logger.debug("Analyzing PE [{}].", pe.getClass().getName());
         for (Field field : fields) {
             logger.trace("Field [{}] is of generic type [{}].", field.getName(), field.getGenericType());

             if (field.getType() == Stream[].class) {
                 logger.debug("Found stream field: {}", field.getGenericType());

                 /* Track what fields have streams with the same event type. */
                 String key = field.getGenericType().toString();
                 typeMap.put(key, field);
             }
         }

         /* Assign streams to stream fields. */
         Multimap<Field, Stream<? extends Event>> assignment = LinkedListMultimap.create();
         for (StreamBuilder<? extends Event> sm : streams) {

             Stream<? extends Event> stream = sm.stream;
             Class<? extends Event> eventType = sm.type;
             String key = Stream.class.getCanonicalName() + "<" + eventType.getCanonicalName() + ">[]";
             if (typeMap.containsKey(key)) {
                 String fieldName;
                 Field field;
                 Collection<Field> streamFields = typeMap.get(key);
                 int numStreamFields = streamFields.size();
                 logger.debug("Found [{}] stream fields for type [{}].", numStreamFields, key);

                 if (numStreamFields > 1) {

                     /*
                      * There is more than one field that can be used for this stream type. To resolve the ambiguity we
                      * need additional information. The app graph should include the name of the field that should be
                      * used to assign this stream. If the name is missing we bail out.
                      */
                     fieldName = sm.fieldName;

                     /* Bail out. */
                     if (fieldName == null) {
                         String msg = String
                                 .format("There are [%d] stream fields in PE [%s]. To assign stream [%s] you need to provide the field name in the application graph using the method withFiled(). See Javadocs for an example.",
                                         numStreamFields, pe.getClass().getName(), stream.getName());
                         logger.error(msg);
                         throw new Exception(msg);
                     }

                     /* Use the provided field name to choose the PE field. */
                     field = pe.getClass().getDeclaredField(fieldName);

                 } else {

                     /*
                      * The easy case, no ambiguity, we don't need an explicit field name to be provided. We have the
                      * field that matches the stream type.
                      */
                     Iterator<Field> iter = streamFields.iterator();
                     field = iter.next(); // Note that numStreamFields == 1, the size of this collection is 1.
                     logger.debug("Using field [{}].", field.getName());
                 }

                 /*
                  * By now, we found the field to use for this stream or we bailed out. We are not ready to finish yet.
                  * There may be more than one stream that needs to be assigned to this field. The stream fields must be
                  * arrays by convention and there may be more than one stream assigned to this fields. For now we create
                  * a multimap from field to streams so we can construct the array in the next step.
                  */
                 assignment.put(field, stream);

             } else {

                 /* We couldn't find a match. Tell user to fix the EDSL code. */
                 String msg = String.format(
                         "There is no stream of type [%s] in PE [%s]. I was unable to assign stream [%s].", key, pe
                                 .getClass().getName(), stream.getName());
                 logger.error(msg);
                 throw new Exception(msg);

             }
         }
         /* Now we construct the array and do the final assignment. */

         Map<Field, Collection<Stream<? extends Event>>> assignmentMap = assignment.asMap();
         for (Map.Entry<Field, Collection<Stream<? extends Event>>> entry : assignmentMap.entrySet()) {
             Field f = entry.getKey();

             int arraySize = entry.getValue().size();
             @SuppressWarnings("unchecked")
             Stream<? extends Event> streamArray[] = (Stream<? extends Event>[]) Array.newInstance(Stream.class,
                     arraySize);
             int i = 0;
             for (Stream<? extends Event> s : entry.getValue()) {
                 streamArray[i++] = s;

                 f.setAccessible(true);
                 f.set(pe, streamArray);
                 logger.debug("Assigned [{}] streams to field [{}].", streamArray.length, f.getName());
             }
         }
     }

     void clearPEState() {
         propertyName = null;
         propertyValue = null;
         processingElement = null;
         peName = null;
         triggerEventType = null;
         triggerTimeUnit = null;
         cacheSize = -1;
         streamBuilder = null;
     }

 }
	/**
	* 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.s4.edsl;

	import java.lang.reflect.Array;
	import java.lang.reflect.Field;
	import java.util.Collection;
	import java.util.Iterator;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.TimeUnit;

	import org.apache.s4.base.Event;
	import org.apache.s4.core.App;
	import org.apache.s4.core.ProcessingElement;
	import org.apache.s4.core.Stream;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.collect.LinkedListMultimap;
	import com.google.common.collect.Multimap;
	import com.google.common.collect.Sets;

	/**
	* Implementation of the S4 embedded domain-specific language (EDSL).
	*
	* <p>
	* To write an app extend this class and define the application graph using a chain of methods as follows:
	*
	* <pre>
	* final public class MyApp extends BuilderS4DSL {
	*
	* protected void onInit() {
	*
	* pe("Consumer").type(ConsumerPE.class).asSingleton().
	* pe("Producer").type(ProducerPE.class).timer().withPeriod(1, TimeUnit.MILLISECONDS).asSingleton().
	* emit(SomeEvent.class).withKey("someKey").to("Consumer").
	* build()
	* }
	* </pre>
	*
	* <p>
	* A few things to notice:
	* <ul>
	* <li>Applications must extend class {@link BuilderS4DSL}
	* <li>The graph definition is implemented in the {@link App#onInit} method which is called by the container when the
	* application is loaded.
	* <li>PEs are defined using strings because they need to be referenced by other parts of the graph. By doing this, we
	* can create the whole application in a single chain of methods.
	* <li>To assign target streams to PE fields additional information may need to be provided using the {@code onField}
	* grammar token when there is an ambiguity. This will happen when a PE has more than one targetStream field with the
	* same {@link Event} type. Use the construct {@code emit(SomeEvent.class).onField("streamFieldName")}. If the PE
	* doesn't have a field named {@code "streamField"} whose stream parameter type is {@code someEvent)} then the parser
	* will fail to build the app.
	* <li>To configure a PE, set property values by chaining any number of {@code prop(name, value)} methods. The name
	* should match a PE field and the value will be parsed using the type of that field.
	* </ul>
	* <p>
	* Grammar:
	*
	* <pre>
	* (pe , type , prop* , (fireOn , afterInterval? , afterNumEvents?)? , (timer, withPeriod)? ,
	* (cache, size , expires? )? , asSingleton? , (emit, onField?,
	* (withKey\|withKeyFinder)?, to )* )+ , build
	* </pre>
	*
	* <p>
	* See the <a href="http://code.google.com/p/diezel">Diezel</a> project for details.
	*
	*/
	public class AppBuilder extends App {

	protected App app = this;

	static final Logger logger = LoggerFactory.getLogger(AppBuilder.class);

	private Multimap<ProcessingElement, StreamBuilder<? extends Event>> pe2stream = LinkedListMultimap.create();
	Set<StreamBuilder<? extends Event>> streamBuilders = Sets.newHashSet();

	/* Variables used to hold values from state to state. */
	ProcessingElement processingElement;
	String peName;
	Class<? extends Event> triggerEventType;
	long triggerInterval = 0;
	TimeUnit triggerTimeUnit;
	int cacheSize;
	StreamBuilder<? extends Event> streamBuilder;
	String propertyName, propertyValue;

	public static AppBuilder getAppBuilder() {
	return new BuilderS4DSL();
	}

	void addProperty(String name, String value) {
	propertyName = name;
	propertyValue = value;
	setField();
	}

	void addPe2Stream(ProcessingElement pe, StreamBuilder<? extends Event> st) {
	pe2stream.put(pe, st);
	}

	App buildApp() {

	/* Stream to PE writing. */
	for (StreamBuilder<? extends Event> sb : streamBuilders) {
	for (String peName : sb.pes) {
	ProcessingElement pe = getPE(peName);
	sb.stream.setPEs(pe);
	}
	}

	/* PE to Stream wiring. */
	Map<ProcessingElement, Collection<StreamBuilder<? extends Event>>> pe2streamMap = pe2stream.asMap();
	for (Map.Entry<ProcessingElement, Collection<StreamBuilder<? extends Event>>> entry : pe2streamMap.entrySet()) {
	ProcessingElement pe = entry.getKey();
	Collection<StreamBuilder<? extends Event>> streams = entry.getValue();

	if (pe != null && streams != null) {
	try {
	setStreamField(pe, streams);
	} catch (Exception e) {
	logger.error("Unable to build app.", e);
	return null;
	}
	}
	}

	return this;
	}

	/**
	* @param peName
	* the peName to set
	*/
	protected void setPeName(String peName) {
	this.peName = peName;
	}

	/*
	* Cannot create an abstract class in Diezel so for now, I just implement the abstract methods here. They need to be
	* overloaded by the app developer.
	*/
	@Override
	protected void onStart() {
	}

	@Override
	protected void onInit() {
	}

	@Override
	protected void onClose() {
	}

	private <T extends ProcessingElement> void setField() {

	logger.debug("Adding property [{}] to PE of type [{}].", propertyName, processingElement.getClass().getName());

	Class<? extends ProcessingElement> type = processingElement.getClass();

	try {
	Field f = type.getDeclaredField(propertyName);
	f.setAccessible(true);
	logger.trace("Type: {}.", f.getType());
	logger.trace("GenericType: {}.", f.getGenericType());

	/* Set the field. */
	if (f.getType().getCanonicalName() == "long") {
	f.setLong(processingElement, Long.parseLong(propertyValue));
	return;
	} else if (f.getType().getCanonicalName() == "int") {
	f.setInt(processingElement, Integer.parseInt(propertyValue));
	return;
	} else if (f.getType().getCanonicalName() == "float") {
	f.setFloat(processingElement, Float.parseFloat(propertyValue));
	return;
	} else if (f.getType().getCanonicalName() == "double") {
	f.setDouble(processingElement, Double.parseDouble(propertyValue));
	return;
	} else if (f.getType().getCanonicalName() == "short") {
	f.setShort(processingElement, Short.parseShort(propertyValue));
	return;
	} else if (f.getType().getCanonicalName() == "byte") {
	f.setByte(processingElement, Byte.parseByte(propertyValue));
	return;
	} else if (f.getType().getCanonicalName() == "boolean") {
	f.setBoolean(processingElement, Boolean.parseBoolean(propertyValue));
	return;
	} else if (f.getType().getCanonicalName() == "char") {
	f.setChar(processingElement, (char) Byte.parseByte(propertyValue));
	return;
	} else if (f.getType().getCanonicalName() == "java.lang.String") {
	f.set(processingElement, propertyValue);
	return;
	}

	logger.error("Unable to set field named [{}] in PE of type [{}].", propertyName, type);
	throw new IllegalArgumentException();

	// production code should handle these exceptions more gracefully
	} catch (NoSuchFieldException e) {
	logger.error("There is no field named [{}] in PE of type [{}].", propertyName, type);
	} catch (Exception e) {
	logger.error("Couldn't set value for field [{}] in PE of type [{}].", propertyName, type);
	}
	}

	/* Set the stream fields in PE classes. Infer the field by checking the stream parameter type <? extends Event>. */
	private void setStreamField(ProcessingElement pe, Collection<StreamBuilder<? extends Event>> streams)
	throws Exception {

	/*
	* Create a map of the stream fields to the corresponding generic type. We will use this info to assign the
	* streams. If the field type matches the stream type and there is no ambiguity, then the assignment is easy. If
	* more than one field has the same type, then then we need to do more work.
	*/
	Field[] fields = pe.getClass().getDeclaredFields();
	Multimap<String, Field> typeMap = LinkedListMultimap.create();
	logger.debug("Analyzing PE [{}].", pe.getClass().getName());
	for (Field field : fields) {
	logger.trace("Field [{}] is of generic type [{}].", field.getName(), field.getGenericType());

	if (field.getType() == Stream[].class) {
	logger.debug("Found stream field: {}", field.getGenericType());

	/* Track what fields have streams with the same event type. */
	String key = field.getGenericType().toString();
	typeMap.put(key, field);
	}
	}

	/* Assign streams to stream fields. */
	Multimap<Field, Stream<? extends Event>> assignment = LinkedListMultimap.create();
	for (StreamBuilder<? extends Event> sm : streams) {

	Stream<? extends Event> stream = sm.stream;
	Class<? extends Event> eventType = sm.type;
	String key = Stream.class.getCanonicalName() + "<" + eventType.getCanonicalName() + ">[]";
	if (typeMap.containsKey(key)) {
	String fieldName;
	Field field;
	Collection<Field> streamFields = typeMap.get(key);
	int numStreamFields = streamFields.size();
	logger.debug("Found [{}] stream fields for type [{}].", numStreamFields, key);

	if (numStreamFields > 1) {

	/*
	* There is more than one field that can be used for this stream type. To resolve the ambiguity we
	* need additional information. The app graph should include the name of the field that should be
	* used to assign this stream. If the name is missing we bail out.
	*/
	fieldName = sm.fieldName;

	/* Bail out. */
	if (fieldName == null) {
	String msg = String
	.format("There are [%d] stream fields in PE [%s]. To assign stream [%s] you need to provide the field name in the application graph using the method withFiled(). See Javadocs for an example.",
	numStreamFields, pe.getClass().getName(), stream.getName());
	logger.error(msg);
	throw new Exception(msg);
	}

	/* Use the provided field name to choose the PE field. */
	field = pe.getClass().getDeclaredField(fieldName);

	} else {

	/*
	* The easy case, no ambiguity, we don't need an explicit field name to be provided. We have the
	* field that matches the stream type.
	*/
	Iterator<Field> iter = streamFields.iterator();
	field = iter.next(); // Note that numStreamFields == 1, the size of this collection is 1.
	logger.debug("Using field [{}].", field.getName());
	}

	/*
	* By now, we found the field to use for this stream or we bailed out. We are not ready to finish yet.
	* There may be more than one stream that needs to be assigned to this field. The stream fields must be
	* arrays by convention and there may be more than one stream assigned to this fields. For now we create
	* a multimap from field to streams so we can construct the array in the next step.
	*/
	assignment.put(field, stream);

	} else {

	/* We couldn't find a match. Tell user to fix the EDSL code. */
	String msg = String.format(
	"There is no stream of type [%s] in PE [%s]. I was unable to assign stream [%s].", key, pe
	.getClass().getName(), stream.getName());
	logger.error(msg);
	throw new Exception(msg);

	}
	}
	/* Now we construct the array and do the final assignment. */

	Map<Field, Collection<Stream<? extends Event>>> assignmentMap = assignment.asMap();
	for (Map.Entry<Field, Collection<Stream<? extends Event>>> entry : assignmentMap.entrySet()) {
	Field f = entry.getKey();

	int arraySize = entry.getValue().size();
	@SuppressWarnings("unchecked")
	Stream<? extends Event> streamArray[] = (Stream<? extends Event>[]) Array.newInstance(Stream.class,
	arraySize);
	int i = 0;
	for (Stream<? extends Event> s : entry.getValue()) {
	streamArray[i++] = s;

	f.setAccessible(true);
	f.set(pe, streamArray);
	logger.debug("Assigned [{}] streams to field [{}].", streamArray.length, f.getName());
	}
	}
	}

	void clearPEState() {
	propertyName = null;
	propertyValue = null;
	processingElement = null;
	peName = null;
	triggerEventType = null;
	triggerTimeUnit = null;
	cacheSize = -1;
	streamBuilder = null;
	}

	}