tMongoDBOutput_main.javajet

<%@ jet
imports="
	org.talend.designer.codegen.config.CodeGeneratorArgument
	org.talend.core.model.process.INode
	org.talend.core.model.process.ElementParameterParser
	org.talend.core.model.process.IConnection
	org.talend.core.model.metadata.IMetadataColumn
	org.talend.core.model.metadata.IMetadataTable
	org.talend.core.model.metadata.types.JavaTypesManager
	org.talend.core.model.metadata.types.JavaType
	org.talend.core.model.process.IConnectionCategory
	org.talend.core.model.metadata.MappingTypeRetriever
	org.talend.core.model.metadata.MetadataTalendType
	java.util.List
	java.util.ArrayList
	java.util.Map
	java.util.HashMap
	java.util.LinkedList
"
%>
<%
  CodeGeneratorArgument codeGenArgument = (CodeGeneratorArgument) argument;
  INode node = (INode)codeGenArgument.getArgument();
  String cid = node.getUniqueName();
  String dataAction = ElementParameterParser.getValue(node,"__DATA_ACTION__");
  boolean dontStoreNulls = new Boolean(ElementParameterParser.getValue(node,"__DONT_STORE_NULL_VALUES__"));

  /*
   * Retrieve internal schema : build columnList
   */
  List<IMetadataColumn> columnList = null;
  List<IMetadataTable> metadatas = node.getMetadataList();
  if(metadatas != null && metadatas.size() > 0)
  {
      IMetadataTable metadata = metadatas.get(0);
      if(metadata != null)
      {
          columnList = metadata.getListColumns();
      }
  }


  /*
   * Retrieve input and output Talend connections
   */
  List<? extends IConnection> inputs = node.getIncomingConnections();
  List<? extends IConnection> outputs = node.getOutgoingSortedConnections();
  String incomingConnName = null;
  String outgoingConnName = null;

  if (inputs.size() > 0)
  {
      IConnection in = inputs.get(0);
      incomingConnName = in.getName();
  }

  if (outputs.size() > 0)
  {
      IConnection out = outputs.get(0);
      outgoingConnName = out.getName();
  }


  /*
   * Create a new document to insert, and fill columns 
   * iterating over each column
   */
  %>
  com.mongodb.BasicDBObject newDoc = new com.mongodb.BasicDBObject();
  com.mongodb.BasicDBObject nullFields = new com.mongodb.BasicDBObject();
  org.bson.types.ObjectId newDocId = null;
  <%
  for(IMetadataColumn colStmt : columnList)
  {
    // Forward input values to output link if any
    if (outgoingConnName != null)
    {
      %>
      <%=outgoingConnName%>.<%=colStmt%> = <%=incomingConnName%>.<%=colStmt%>;
      <%
    }
    if ("_id".equals(colStmt.getLabel()))
    {
    %>
      newDocId = (org.bson.types.ObjectId) <%=incomingConnName%>.<%=colStmt%>;
      <%
      if ("UPDATE".equals(dataAction) || "UPDATE_OR_INSERT".equals(dataAction)) {
        continue;
      }
    }
    
    if (dontStoreNulls && colStmt.isNullable()) {
	    %>
	      if (<%=incomingConnName%>.<%=colStmt%> != null) {
	      	newDoc.put("<%=colStmt%>", <%=incomingConnName%>.<%=colStmt%>);
	      } else {
	      	nullFields.put("<%=colStmt%>", "1");
	      }
	  	<%
	} else {
		%>
	      	newDoc.put("<%=colStmt%>", <%=incomingConnName%>.<%=colStmt%>);
	  	<%
	}
  }
  if ("INSERT".equals(dataAction)) {
%>
      coll_<%=cid%>.insert(newDoc);
<%
  } else if ("INSERT_OR_REPLACE".equals(dataAction)) {
%>
      coll_<%=cid%>.save(newDoc);
<%
  } else if ("UPDATE".equals(dataAction) || "UPDATE_OR_INSERT".equals(dataAction)) {
%>
      com.mongodb.BasicDBObject selector = new com.mongodb.BasicDBObject("_id", newDocId);
      com.mongodb.BasicDBObject updateOp = new com.mongodb.BasicDBObject("$set", newDoc);
      boolean upsert = <%="UPDATE_OR_INSERT".equals(dataAction)%>;
      coll_<%=cid%>.update(selector, updateOp, upsert, false);
      
      <% if (dontStoreNulls) { %>
	      // Unset any values that were null
	      if (!nullFields.isEmpty()) {
	      	updateOp = new com.mongodb.BasicDBObject("$unset", nullFields);
	      	coll_<%=cid%>.update(selector, updateOp, false, false);
	      }
<%
	  }
  } else if ("DELETE".equals(dataAction)) {
%>
      coll_<%=cid%>.remove(newDoc);
<%
  }
%>
++nb_line_<%=cid %>;