engine/src/main/java/com/datatorrent/stram/codec/LogicalPlanSerializer.java - apex-core - 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 com.datatorrent.stram.codec;

 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;

 import javax.ws.rs.Produces;
 import javax.ws.rs.ext.Provider;

 import org.codehaus.jackson.JsonGenerator;
 import org.codehaus.jackson.annotate.JsonTypeInfo;
 import org.codehaus.jackson.map.JsonSerializer;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.map.ObjectMapper.DefaultTypeResolverBuilder;
 import org.codehaus.jackson.map.ObjectMapper.DefaultTyping;
 import org.codehaus.jackson.map.SerializationConfig;
 import org.codehaus.jackson.map.SerializerProvider;
 import org.codehaus.jackson.map.jsontype.impl.StdTypeResolverBuilder;
 import org.codehaus.jackson.type.JavaType;
 import org.codehaus.jettison.json.JSONArray;
 import org.codehaus.jettison.json.JSONException;
 import org.codehaus.jettison.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import org.apache.commons.beanutils.BeanMap;
 import org.apache.commons.configuration.PropertiesConfiguration;

 import com.datatorrent.api.Attribute;
 import com.datatorrent.api.Context;
 import com.datatorrent.api.DAG.Locality;
 import com.datatorrent.api.Operator;
 import com.datatorrent.api.Operator.InputPort;
 import com.datatorrent.api.Operator.OutputPort;
 import com.datatorrent.common.util.ObjectMapperString;
 import com.datatorrent.stram.plan.logical.LogicalPlan;
 import com.datatorrent.stram.plan.logical.LogicalPlan.InputPortMeta;
 import com.datatorrent.stram.plan.logical.LogicalPlan.OperatorMeta;
 import com.datatorrent.stram.plan.logical.LogicalPlan.OutputPortMeta;
 import com.datatorrent.stram.plan.logical.LogicalPlan.StreamMeta;
 import com.datatorrent.stram.plan.logical.LogicalPlanConfiguration;
 import com.datatorrent.stram.plan.logical.Operators;
 import com.datatorrent.stram.plan.logical.Operators.PortContextPair;

 /**
  * <p>LogicalPlanSerializer class.</p>
  *
  * @since 0.3.2
  */
 @Provider
 @Produces("application/json")
 public class LogicalPlanSerializer extends JsonSerializer<LogicalPlan>
 {
   private static class PropertyTypeResolverBuilder extends DefaultTypeResolverBuilder
   {
     PropertyTypeResolverBuilder()
     {
       super(DefaultTyping.NON_FINAL);
     }

     @Override
     public boolean useForType(JavaType t)
     {
       if (t.getRawClass() == Object.class) {
         return true;
       }
       if (t.getRawClass().getName().startsWith("java.")) {
         return false;
       }
       if (t.isArrayType()) {
         return false;
       }
       return super.useForType(t);
     }

   }

   private static final Logger LOG = LoggerFactory.getLogger(LogicalPlanSerializer.class);

   /**
    *
    * @param dag
    * @return
    */
   public static Map<String, Object> convertToMap(LogicalPlan dag, boolean includeModules)
   {
     HashMap<String, Object> result = new HashMap<>();
     ArrayList<Object> operatorArray = new ArrayList<>();
     ArrayList<Object> streamMap = new ArrayList<>();
     //result.put("applicationName", appConfig.getName());
     result.put("operators", operatorArray);
     result.put("streams", streamMap);
     //LogicalPlan dag = StramAppLauncher.prepareDAG(appConfig, StreamingApplication.LAUNCHMODE_YARN);
     //
     // should we put the DAGContext info here?

     Map<String, Object> dagAttrs = new HashMap<>();
     for (Map.Entry<Attribute<Object>, Object> e :
         Attribute.AttributeMap.AttributeInitializer.getAllAttributes(dag, Context.DAGContext.class).entrySet()) {
       dagAttrs.put(e.getKey().getSimpleName(), e.getValue());
     }
     result.put("attributes", dagAttrs);

     Collection<OperatorMeta> allOperators = dag.getAllOperators();
     ObjectMapper propertyObjectMapper = new ObjectMapper();
     propertyObjectMapper.configure(JsonGenerator.Feature.WRITE_NUMBERS_AS_STRINGS, true);
     propertyObjectMapper.configure(SerializationConfig.Feature.FAIL_ON_EMPTY_BEANS, false);

     StdTypeResolverBuilder typer = new PropertyTypeResolverBuilder();
     typer.init(JsonTypeInfo.Id.CLASS, null);
     typer = typer.inclusion(JsonTypeInfo.As.PROPERTY);
     propertyObjectMapper.setDefaultTyping(typer);

     for (OperatorMeta operatorMeta : allOperators) {
       HashMap<String, Object> operatorDetailMap = new HashMap<>();
       ArrayList<Map<String, Object>> portList = new ArrayList<>();
       Map<String, Object> attributeMap = new HashMap<>();

       String operatorName = operatorMeta.getName();
       operatorArray.add(operatorDetailMap);
       operatorDetailMap.put("name", operatorName);
       operatorDetailMap.put("ports", portList);
       operatorDetailMap.put("class", operatorMeta.getOperator().getClass().getName());
       operatorDetailMap.put("attributes", attributeMap);
       Map<Attribute<Object>, Object> rawAttributes = Attribute.AttributeMap.AttributeInitializer.getAllAttributes(operatorMeta, Context.OperatorContext.class);
       for (Map.Entry<Attribute<Object>, Object> entry : rawAttributes.entrySet()) {
         attributeMap.put(entry.getKey().getSimpleName(), entry.getValue());
       }

       ObjectMapperString str;

       try {
         str = new ObjectMapperString(propertyObjectMapper.writeValueAsString(operatorMeta.getOperator()));
       } catch (Throwable ex) {
         LOG.error("Got exception when trying to get properties for operator {}", operatorMeta.getName(), ex);
         str = null;
       }
       operatorDetailMap.put("properties", str);

       Operators.PortMappingDescriptor pmd = new Operators.PortMappingDescriptor();
       Operators.describe(operatorMeta.getOperator(), pmd);
       for (Map.Entry<String, PortContextPair<InputPort<?>>> entry : pmd.inputPorts.entrySet()) {
         HashMap<String, Object> portDetailMap = new HashMap<>();
         HashMap<String, Object> portAttributeMap = new HashMap<>();
         InputPortMeta portMeta = operatorMeta.getMeta(entry.getValue().component);
         String portName = portMeta.getPortName();
         portDetailMap.put("name", portName);
         portDetailMap.put("type", "input");
         portDetailMap.put("attributes", portAttributeMap);
         rawAttributes = Attribute.AttributeMap.AttributeInitializer.getAllAttributes(portMeta, Context.PortContext.class);
         for (Map.Entry<Attribute<Object>, Object> attEntry : rawAttributes.entrySet()) {
           portAttributeMap.put(attEntry.getKey().getSimpleName(), attEntry.getValue());
         }
         portList.add(portDetailMap);
       }
       for (Map.Entry<String, PortContextPair<OutputPort<?>>> entry : pmd.outputPorts.entrySet()) {
         HashMap<String, Object> portDetailMap = new HashMap<>();
         HashMap<String, Object> portAttributeMap = new HashMap<>();
         OutputPortMeta portMeta = operatorMeta.getMeta(entry.getValue().component);
         String portName = portMeta.getPortName();
         portDetailMap.put("name", portName);
         portDetailMap.put("type", "output");
         portDetailMap.put("attributes", portAttributeMap);
         rawAttributes = Attribute.AttributeMap.AttributeInitializer.getAllAttributes(portMeta, Context.PortContext.class);
         for (Map.Entry<Attribute<Object>, Object> attEntry : rawAttributes.entrySet()) {
           portAttributeMap.put(attEntry.getKey().getSimpleName(), attEntry.getValue());
         }
         portList.add(portDetailMap);
       }
     }
     Collection<StreamMeta> allStreams = dag.getAllStreams();

     for (StreamMeta streamMeta : allStreams) {
       HashMap<String, Object> streamDetailMap = new HashMap<>();
       String streamName = streamMeta.getName();
       streamMap.add(streamDetailMap);
       String sourcePortName = streamMeta.getSource().getPortName();
       OperatorMeta operatorMeta = streamMeta.getSource().getOperatorMeta();
       HashMap<String, Object> sourcePortDetailMap = new HashMap<>();
       sourcePortDetailMap.put("operatorName", operatorMeta.getName());
       sourcePortDetailMap.put("portName", sourcePortName);
       streamDetailMap.put("name", streamName);
       streamDetailMap.put("source", sourcePortDetailMap);
       List<InputPortMeta> sinks = streamMeta.getSinks();
       ArrayList<HashMap<String, Object>> sinkPortList = new ArrayList<>();
       for (InputPortMeta sinkPort : sinks) {
         HashMap<String, Object> sinkPortDetailMap = new HashMap<>();
         sinkPortDetailMap.put("operatorName", sinkPort.getOperatorWrapper().getName());
         sinkPortDetailMap.put("portName", sinkPort.getPortName());
         sinkPortList.add(sinkPortDetailMap);
       }
       streamDetailMap.put("sinks", sinkPortList);
       if (streamMeta.getLocality() != null) {
         streamDetailMap.put("locality", streamMeta.getLocality().name());
       }
     }

     if (includeModules) {
       ArrayList<Map<String, Object>> modulesArray = new ArrayList<>();
       result.put("modules", modulesArray);
       for (LogicalPlan.ModuleMeta meta : dag.getAllModules()) {
         modulesArray.add(getLogicalModuleDetails(dag, meta));
       }
     }

     return result;
   }

   public static PropertiesConfiguration convertToProperties(LogicalPlan dag)
   {
     PropertiesConfiguration props = new PropertiesConfiguration();
     Collection<OperatorMeta> allOperators = dag.getAllOperators();

     for (OperatorMeta operatorMeta : allOperators) {
       String operatorKey = LogicalPlanConfiguration.OPERATOR_PREFIX + operatorMeta.getName();
       Operator operator = operatorMeta.getOperator();
       props.setProperty(operatorKey + "." + LogicalPlanConfiguration.OPERATOR_CLASSNAME, operator.getClass().getName());
       BeanMap operatorProperties = LogicalPlanConfiguration.getObjectProperties(operator);
       @SuppressWarnings("rawtypes")
       Iterator entryIterator = operatorProperties.entryIterator();
       while (entryIterator.hasNext()) {
         try {
           @SuppressWarnings("unchecked")
           Map.Entry<String, Object> entry = (Map.Entry<String, Object>)entryIterator.next();
           if (!entry.getKey().equals("class") && !entry.getKey().equals("name") && entry.getValue() != null) {
             props.setProperty(operatorKey + "." + entry.getKey(), entry.getValue());
           }
         } catch (Exception ex) {
           LOG.warn("Error trying to get a property of operator {}", operatorMeta.getName(), ex);
         }
       }
     }
     Collection<StreamMeta> allStreams = dag.getAllStreams();

     for (StreamMeta streamMeta : allStreams) {
       String streamKey = LogicalPlanConfiguration.STREAM_PREFIX + streamMeta.getName();
       OutputPortMeta source = streamMeta.getSource();
       List<InputPortMeta> sinks = streamMeta.getSinks();
       props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_SOURCE, source.getOperatorMeta().getName() + "." + source.getPortName());
       String sinksValue = "";
       for (InputPortMeta sink : sinks) {
         if (!sinksValue.isEmpty()) {
           sinksValue += ",";
         }
         sinksValue += sink.getOperatorWrapper().getName() + "." + sink.getPortName();
       }
       props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_SINKS, sinksValue);
       if (streamMeta.getLocality() != null) {
         props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_LOCALITY, streamMeta.getLocality().name());
       }
     }

     // TBD: Attributes

     return props;
   }

   public static PropertiesConfiguration convertToProperties(JSONObject json) throws JSONException
   {
     PropertiesConfiguration props = new PropertiesConfiguration();
     JSONArray allOperators = json.getJSONArray("operators");
     JSONArray allStreams = json.getJSONArray("streams");

     for (int j = 0; j < allOperators.length(); j++) {
       JSONObject operatorDetail = allOperators.getJSONObject(j);
       String operatorName = operatorDetail.getString("name");
       String operatorKey = LogicalPlanConfiguration.OPERATOR_PREFIX + operatorName;
       props.setProperty(operatorKey + ".classname", operatorDetail.getString("class"));
       JSONObject properties = operatorDetail.optJSONObject("properties");
       if (properties != null) {
         Iterator<String> iter2 = properties.keys();
         while (iter2.hasNext()) {
           String propertyName = iter2.next();
           if (!propertyName.equals("name") && !propertyName.equals("class") && properties.opt(propertyName) != null) {
             JSONArray list = properties.optJSONArray(propertyName);
             String value = "";
             if (list != null) {
               for (int i = 0; i < list.length(); i++) {
                 if (i != 0) {
                   value += ",";
                 }
                 value += list.get(i).toString();
               }
               props.setProperty(operatorKey + "." + propertyName, value);
             } else {
               props.setProperty(operatorKey + "." + propertyName, properties.get(propertyName));
             }
           }
         }
       }
     }

     for (int j = 0; j < allStreams.length(); j++) {
       JSONObject streamDetail = allStreams.getJSONObject(j);
       String streamName = streamDetail.getString("name");
       String streamKey = LogicalPlanConfiguration.STREAM_PREFIX + streamName;
       JSONObject sourceDetail = streamDetail.getJSONObject("source");
       JSONArray sinksList = streamDetail.getJSONArray("sinks");

       props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_SOURCE, sourceDetail.getString("operatorName") + "." + sourceDetail.getString("portName"));
       String sinksValue = "";
       for (int i = 0; i < sinksList.length(); i++) {
         if (!sinksValue.isEmpty()) {
           sinksValue += ",";
         }
         sinksValue += sinksList.getJSONObject(i).getString("operatorName") + "." + sinksList.getJSONObject(i).getString("portName");
       }
       props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_SINKS, sinksValue);
       String locality = streamDetail.optString("locality", null);
       if (locality != null) {
         props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_LOCALITY, Locality.valueOf(locality));
       }
     }

     // TBD: Attributes

     return props;
   }

   public static JSONObject convertToJsonObject(LogicalPlan dag)
   {
     return new JSONObject(convertToMap(dag, false));
   }

   @Override
   public void serialize(LogicalPlan dag, JsonGenerator jg, SerializerProvider sp) throws IOException
   {
     jg.writeObject(convertToMap(dag, false));
   }

   /**
    * Return information about operators and inner modules of a module.
    *
    * @param dag        top level DAG
    * @param moduleMeta module information. DAG within module is used for constructing response.
    * @return
    */
   private static Map<String, Object> getLogicalModuleDetails(LogicalPlan dag, LogicalPlan.ModuleMeta moduleMeta)
   {
     Map<String, Object> moduleDetailMap = new HashMap<>();
     ArrayList<String> operatorArray = new ArrayList<>();
     moduleDetailMap.put("name", moduleMeta.getName());
     moduleDetailMap.put("className", moduleMeta.getGenericOperator().getClass().getName());

     moduleDetailMap.put("operators", operatorArray);
     for (OperatorMeta operatorMeta : moduleMeta.getDag().getAllOperators()) {
       if (operatorMeta.getModuleName() == null) {
         String fullName = moduleMeta.getFullName() + LogicalPlan.MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
         operatorArray.add(fullName);
       }
     }
     ArrayList<Map<String, Object>> modulesArray = new ArrayList<>();
     moduleDetailMap.put("modules", modulesArray);
     for (LogicalPlan.ModuleMeta meta : moduleMeta.getDag().getAllModules()) {
       modulesArray.add(getLogicalModuleDetails(dag, meta));
     }
     return moduleDetailMap;
   }

 }
	/**
	* 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 com.datatorrent.stram.codec;

	import java.io.IOException;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.HashMap;
	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;

	import javax.ws.rs.Produces;
	import javax.ws.rs.ext.Provider;

	import org.codehaus.jackson.JsonGenerator;
	import org.codehaus.jackson.annotate.JsonTypeInfo;
	import org.codehaus.jackson.map.JsonSerializer;
	import org.codehaus.jackson.map.ObjectMapper;
	import org.codehaus.jackson.map.ObjectMapper.DefaultTypeResolverBuilder;
	import org.codehaus.jackson.map.ObjectMapper.DefaultTyping;
	import org.codehaus.jackson.map.SerializationConfig;
	import org.codehaus.jackson.map.SerializerProvider;
	import org.codehaus.jackson.map.jsontype.impl.StdTypeResolverBuilder;
	import org.codehaus.jackson.type.JavaType;
	import org.codehaus.jettison.json.JSONArray;
	import org.codehaus.jettison.json.JSONException;
	import org.codehaus.jettison.json.JSONObject;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import org.apache.commons.beanutils.BeanMap;
	import org.apache.commons.configuration.PropertiesConfiguration;

	import com.datatorrent.api.Attribute;
	import com.datatorrent.api.Context;
	import com.datatorrent.api.DAG.Locality;
	import com.datatorrent.api.Operator;
	import com.datatorrent.api.Operator.InputPort;
	import com.datatorrent.api.Operator.OutputPort;
	import com.datatorrent.common.util.ObjectMapperString;
	import com.datatorrent.stram.plan.logical.LogicalPlan;
	import com.datatorrent.stram.plan.logical.LogicalPlan.InputPortMeta;
	import com.datatorrent.stram.plan.logical.LogicalPlan.OperatorMeta;
	import com.datatorrent.stram.plan.logical.LogicalPlan.OutputPortMeta;
	import com.datatorrent.stram.plan.logical.LogicalPlan.StreamMeta;
	import com.datatorrent.stram.plan.logical.LogicalPlanConfiguration;
	import com.datatorrent.stram.plan.logical.Operators;
	import com.datatorrent.stram.plan.logical.Operators.PortContextPair;

	/**
	* <p>LogicalPlanSerializer class.</p>
	*
	* @since 0.3.2
	*/
	@Provider
	@Produces("application/json")
	public class LogicalPlanSerializer extends JsonSerializer<LogicalPlan>
	{
	private static class PropertyTypeResolverBuilder extends DefaultTypeResolverBuilder
	{
	PropertyTypeResolverBuilder()
	{
	super(DefaultTyping.NON_FINAL);
	}

	@Override
	public boolean useForType(JavaType t)
	{
	if (t.getRawClass() == Object.class) {
	return true;
	}
	if (t.getRawClass().getName().startsWith("java.")) {
	return false;
	}
	if (t.isArrayType()) {
	return false;
	}
	return super.useForType(t);
	}

	}

	private static final Logger LOG = LoggerFactory.getLogger(LogicalPlanSerializer.class);

	/**
	*
	* @param dag
	* @return
	*/
	public static Map<String, Object> convertToMap(LogicalPlan dag, boolean includeModules)
	{
	HashMap<String, Object> result = new HashMap<>();
	ArrayList<Object> operatorArray = new ArrayList<>();
	ArrayList<Object> streamMap = new ArrayList<>();
	//result.put("applicationName", appConfig.getName());
	result.put("operators", operatorArray);
	result.put("streams", streamMap);
	//LogicalPlan dag = StramAppLauncher.prepareDAG(appConfig, StreamingApplication.LAUNCHMODE_YARN);
	//
	// should we put the DAGContext info here?

	Map<String, Object> dagAttrs = new HashMap<>();
	for (Map.Entry<Attribute<Object>, Object> e :
	Attribute.AttributeMap.AttributeInitializer.getAllAttributes(dag, Context.DAGContext.class).entrySet()) {
	dagAttrs.put(e.getKey().getSimpleName(), e.getValue());
	}
	result.put("attributes", dagAttrs);

	Collection<OperatorMeta> allOperators = dag.getAllOperators();
	ObjectMapper propertyObjectMapper = new ObjectMapper();
	propertyObjectMapper.configure(JsonGenerator.Feature.WRITE_NUMBERS_AS_STRINGS, true);
	propertyObjectMapper.configure(SerializationConfig.Feature.FAIL_ON_EMPTY_BEANS, false);

	StdTypeResolverBuilder typer = new PropertyTypeResolverBuilder();
	typer.init(JsonTypeInfo.Id.CLASS, null);
	typer = typer.inclusion(JsonTypeInfo.As.PROPERTY);
	propertyObjectMapper.setDefaultTyping(typer);

	for (OperatorMeta operatorMeta : allOperators) {
	HashMap<String, Object> operatorDetailMap = new HashMap<>();
	ArrayList<Map<String, Object>> portList = new ArrayList<>();
	Map<String, Object> attributeMap = new HashMap<>();

	String operatorName = operatorMeta.getName();
	operatorArray.add(operatorDetailMap);
	operatorDetailMap.put("name", operatorName);
	operatorDetailMap.put("ports", portList);
	operatorDetailMap.put("class", operatorMeta.getOperator().getClass().getName());
	operatorDetailMap.put("attributes", attributeMap);
	Map<Attribute<Object>, Object> rawAttributes = Attribute.AttributeMap.AttributeInitializer.getAllAttributes(operatorMeta, Context.OperatorContext.class);
	for (Map.Entry<Attribute<Object>, Object> entry : rawAttributes.entrySet()) {
	attributeMap.put(entry.getKey().getSimpleName(), entry.getValue());
	}

	ObjectMapperString str;

	try {
	str = new ObjectMapperString(propertyObjectMapper.writeValueAsString(operatorMeta.getOperator()));
	} catch (Throwable ex) {
	LOG.error("Got exception when trying to get properties for operator {}", operatorMeta.getName(), ex);
	str = null;
	}
	operatorDetailMap.put("properties", str);

	Operators.PortMappingDescriptor pmd = new Operators.PortMappingDescriptor();
	Operators.describe(operatorMeta.getOperator(), pmd);
	for (Map.Entry<String, PortContextPair<InputPort<?>>> entry : pmd.inputPorts.entrySet()) {
	HashMap<String, Object> portDetailMap = new HashMap<>();
	HashMap<String, Object> portAttributeMap = new HashMap<>();
	InputPortMeta portMeta = operatorMeta.getMeta(entry.getValue().component);
	String portName = portMeta.getPortName();
	portDetailMap.put("name", portName);
	portDetailMap.put("type", "input");
	portDetailMap.put("attributes", portAttributeMap);
	rawAttributes = Attribute.AttributeMap.AttributeInitializer.getAllAttributes(portMeta, Context.PortContext.class);
	for (Map.Entry<Attribute<Object>, Object> attEntry : rawAttributes.entrySet()) {
	portAttributeMap.put(attEntry.getKey().getSimpleName(), attEntry.getValue());
	}
	portList.add(portDetailMap);
	}
	for (Map.Entry<String, PortContextPair<OutputPort<?>>> entry : pmd.outputPorts.entrySet()) {
	HashMap<String, Object> portDetailMap = new HashMap<>();
	HashMap<String, Object> portAttributeMap = new HashMap<>();
	OutputPortMeta portMeta = operatorMeta.getMeta(entry.getValue().component);
	String portName = portMeta.getPortName();
	portDetailMap.put("name", portName);
	portDetailMap.put("type", "output");
	portDetailMap.put("attributes", portAttributeMap);
	rawAttributes = Attribute.AttributeMap.AttributeInitializer.getAllAttributes(portMeta, Context.PortContext.class);
	for (Map.Entry<Attribute<Object>, Object> attEntry : rawAttributes.entrySet()) {
	portAttributeMap.put(attEntry.getKey().getSimpleName(), attEntry.getValue());
	}
	portList.add(portDetailMap);
	}
	}
	Collection<StreamMeta> allStreams = dag.getAllStreams();

	for (StreamMeta streamMeta : allStreams) {
	HashMap<String, Object> streamDetailMap = new HashMap<>();
	String streamName = streamMeta.getName();
	streamMap.add(streamDetailMap);
	String sourcePortName = streamMeta.getSource().getPortName();
	OperatorMeta operatorMeta = streamMeta.getSource().getOperatorMeta();
	HashMap<String, Object> sourcePortDetailMap = new HashMap<>();
	sourcePortDetailMap.put("operatorName", operatorMeta.getName());
	sourcePortDetailMap.put("portName", sourcePortName);
	streamDetailMap.put("name", streamName);
	streamDetailMap.put("source", sourcePortDetailMap);
	List<InputPortMeta> sinks = streamMeta.getSinks();
	ArrayList<HashMap<String, Object>> sinkPortList = new ArrayList<>();
	for (InputPortMeta sinkPort : sinks) {
	HashMap<String, Object> sinkPortDetailMap = new HashMap<>();
	sinkPortDetailMap.put("operatorName", sinkPort.getOperatorWrapper().getName());
	sinkPortDetailMap.put("portName", sinkPort.getPortName());
	sinkPortList.add(sinkPortDetailMap);
	}
	streamDetailMap.put("sinks", sinkPortList);
	if (streamMeta.getLocality() != null) {
	streamDetailMap.put("locality", streamMeta.getLocality().name());
	}
	}

	if (includeModules) {
	ArrayList<Map<String, Object>> modulesArray = new ArrayList<>();
	result.put("modules", modulesArray);
	for (LogicalPlan.ModuleMeta meta : dag.getAllModules()) {
	modulesArray.add(getLogicalModuleDetails(dag, meta));
	}
	}

	return result;
	}

	public static PropertiesConfiguration convertToProperties(LogicalPlan dag)
	{
	PropertiesConfiguration props = new PropertiesConfiguration();
	Collection<OperatorMeta> allOperators = dag.getAllOperators();

	for (OperatorMeta operatorMeta : allOperators) {
	String operatorKey = LogicalPlanConfiguration.OPERATOR_PREFIX + operatorMeta.getName();
	Operator operator = operatorMeta.getOperator();
	props.setProperty(operatorKey + "." + LogicalPlanConfiguration.OPERATOR_CLASSNAME, operator.getClass().getName());
	BeanMap operatorProperties = LogicalPlanConfiguration.getObjectProperties(operator);
	@SuppressWarnings("rawtypes")
	Iterator entryIterator = operatorProperties.entryIterator();
	while (entryIterator.hasNext()) {
	try {
	@SuppressWarnings("unchecked")
	Map.Entry<String, Object> entry = (Map.Entry<String, Object>)entryIterator.next();
	if (!entry.getKey().equals("class") && !entry.getKey().equals("name") && entry.getValue() != null) {
	props.setProperty(operatorKey + "." + entry.getKey(), entry.getValue());
	}
	} catch (Exception ex) {
	LOG.warn("Error trying to get a property of operator {}", operatorMeta.getName(), ex);
	}
	}
	}
	Collection<StreamMeta> allStreams = dag.getAllStreams();

	for (StreamMeta streamMeta : allStreams) {
	String streamKey = LogicalPlanConfiguration.STREAM_PREFIX + streamMeta.getName();
	OutputPortMeta source = streamMeta.getSource();
	List<InputPortMeta> sinks = streamMeta.getSinks();
	props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_SOURCE, source.getOperatorMeta().getName() + "." + source.getPortName());
	String sinksValue = "";
	for (InputPortMeta sink : sinks) {
	if (!sinksValue.isEmpty()) {
	sinksValue += ",";
	}
	sinksValue += sink.getOperatorWrapper().getName() + "." + sink.getPortName();
	}
	props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_SINKS, sinksValue);
	if (streamMeta.getLocality() != null) {
	props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_LOCALITY, streamMeta.getLocality().name());
	}
	}

	// TBD: Attributes

	return props;
	}

	public static PropertiesConfiguration convertToProperties(JSONObject json) throws JSONException
	{
	PropertiesConfiguration props = new PropertiesConfiguration();
	JSONArray allOperators = json.getJSONArray("operators");
	JSONArray allStreams = json.getJSONArray("streams");

	for (int j = 0; j < allOperators.length(); j++) {
	JSONObject operatorDetail = allOperators.getJSONObject(j);
	String operatorName = operatorDetail.getString("name");
	String operatorKey = LogicalPlanConfiguration.OPERATOR_PREFIX + operatorName;
	props.setProperty(operatorKey + ".classname", operatorDetail.getString("class"));
	JSONObject properties = operatorDetail.optJSONObject("properties");
	if (properties != null) {
	Iterator<String> iter2 = properties.keys();
	while (iter2.hasNext()) {
	String propertyName = iter2.next();
	if (!propertyName.equals("name") && !propertyName.equals("class") && properties.opt(propertyName) != null) {
	JSONArray list = properties.optJSONArray(propertyName);
	String value = "";
	if (list != null) {
	for (int i = 0; i < list.length(); i++) {
	if (i != 0) {
	value += ",";
	}
	value += list.get(i).toString();
	}
	props.setProperty(operatorKey + "." + propertyName, value);
	} else {
	props.setProperty(operatorKey + "." + propertyName, properties.get(propertyName));
	}
	}
	}
	}
	}

	for (int j = 0; j < allStreams.length(); j++) {
	JSONObject streamDetail = allStreams.getJSONObject(j);
	String streamName = streamDetail.getString("name");
	String streamKey = LogicalPlanConfiguration.STREAM_PREFIX + streamName;
	JSONObject sourceDetail = streamDetail.getJSONObject("source");
	JSONArray sinksList = streamDetail.getJSONArray("sinks");

	props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_SOURCE, sourceDetail.getString("operatorName") + "." + sourceDetail.getString("portName"));
	String sinksValue = "";
	for (int i = 0; i < sinksList.length(); i++) {
	if (!sinksValue.isEmpty()) {
	sinksValue += ",";
	}
	sinksValue += sinksList.getJSONObject(i).getString("operatorName") + "." + sinksList.getJSONObject(i).getString("portName");
	}
	props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_SINKS, sinksValue);
	String locality = streamDetail.optString("locality", null);
	if (locality != null) {
	props.setProperty(streamKey + "." + LogicalPlanConfiguration.STREAM_LOCALITY, Locality.valueOf(locality));
	}
	}

	// TBD: Attributes

	return props;
	}

	public static JSONObject convertToJsonObject(LogicalPlan dag)
	{
	return new JSONObject(convertToMap(dag, false));
	}

	@Override
	public void serialize(LogicalPlan dag, JsonGenerator jg, SerializerProvider sp) throws IOException
	{
	jg.writeObject(convertToMap(dag, false));
	}

	/**
	* Return information about operators and inner modules of a module.
	*
	* @param dag top level DAG
	* @param moduleMeta module information. DAG within module is used for constructing response.
	* @return
	*/
	private static Map<String, Object> getLogicalModuleDetails(LogicalPlan dag, LogicalPlan.ModuleMeta moduleMeta)
	{
	Map<String, Object> moduleDetailMap = new HashMap<>();
	ArrayList<String> operatorArray = new ArrayList<>();
	moduleDetailMap.put("name", moduleMeta.getName());
	moduleDetailMap.put("className", moduleMeta.getGenericOperator().getClass().getName());

	moduleDetailMap.put("operators", operatorArray);
	for (OperatorMeta operatorMeta : moduleMeta.getDag().getAllOperators()) {
	if (operatorMeta.getModuleName() == null) {
	String fullName = moduleMeta.getFullName() + LogicalPlan.MODULE_NAMESPACE_SEPARATOR + operatorMeta.getName();
	operatorArray.add(fullName);
	}
	}
	ArrayList<Map<String, Object>> modulesArray = new ArrayList<>();
	moduleDetailMap.put("modules", modulesArray);
	for (LogicalPlan.ModuleMeta meta : moduleMeta.getDag().getAllModules()) {
	modulesArray.add(getLogicalModuleDetails(dag, meta));
	}
	return moduleDetailMap;
	}

	}