tMongoDBXMLOutput_begin.javajet

<%@ jet 
	imports="
		org.talend.core.model.process.INode 
		org.talend.core.model.process.ElementParameterParser 
		org.talend.core.model.metadata.IMetadataTable 
		org.talend.designer.codegen.config.CodeGeneratorArgument
		java.util.List
		java.util.Map
	" 
%>
<% 
	CodeGeneratorArgument codeGenArgument = (CodeGeneratorArgument) argument;
	INode node = (INode)codeGenArgument.getArgument();
  	String cid = node.getUniqueName();

    String dbhost   = ElementParameterParser.getValue(node, "__HOST__");
    String dbname   = ElementParameterParser.getValue(node, "__DATABASE__");
    boolean authentication="true".equalsIgnoreCase(ElementParameterParser.getValue(node, "__REQUIRED_AUTHENTICATION__"));
    String authenticationMechanism = ElementParameterParser.getValue(node, "__AUTHENTICATION_MECHANISM__");
    String krbUserPrincipal = ElementParameterParser.getValue(node, "__KRB_USER_PRINCIPAL__");
    String krbRealm = ElementParameterParser.getValue(node, "__KRB_REALM__");
    String krbKdc = ElementParameterParser.getValue(node, "__KRB_KDC__");
    String dbuser   = ElementParameterParser.getValue(node, "__USERNAME__");
    String dbpass   = ElementParameterParser.getValue(node, "__PASSWORD__");
    String mongoPort = ElementParameterParser.getValue(node, "__PORT__");
    String dbport = mongoPort.startsWith("context.") ? "Integer.valueOf(" + mongoPort + ").intValue()" : mongoPort.replace("\"", "");
    String collection=ElementParameterParser.getValue(node, "__COLLECTION__");
    boolean dropExistCollection="true".equalsIgnoreCase(ElementParameterParser.getValue(node,"__DROP_COLLECTION_CREATE__"));
    boolean useExistingConnection = "true".equalsIgnoreCase(ElementParameterParser.getValue(node,"__USE_EXISTING_CONNECTION__"));
    String dbversion = ElementParameterParser.getValue(node, "__DB_VERSION__");
    
    List<Map<String, String>> mapping = (List<Map<String,String>>)ElementParameterParser.getObjectValueXML(node, "__MAPPING__");
    boolean useReplicaSet="true".equalsIgnoreCase(ElementParameterParser.getValue(node, "__USE_REPLICA_SET__"));

    boolean setWriteConcern = "true".equalsIgnoreCase(ElementParameterParser.getValue(node, "__SET_WRITE_CONCERN__"));
    String writeConcern = ElementParameterParser.getValue(node, "__WRITE_CONCERN__");

    boolean setBulkWrite = "true".equalsIgnoreCase(ElementParameterParser.getValue(node, "__SET_BULK_WRITE__"));
    String bulkWriteOperationSize = ElementParameterParser.getValue(node, "__BULK_WRITE_SIZE__");
    String bulkWriteType = ElementParameterParser.getValue(node, "__BULK_WRITE_TYPE__");
    // BulkWrite is only supportted on Mongo 2.6+ drivers
    final boolean bulkWrite = setBulkWrite & (!"MONGODB_2_5_X".equalsIgnoreCase(dbversion));

    boolean isLog4jEnabled = ("true").equals(ElementParameterParser.getValue(node.getProcess(), "__LOG4J_ACTIVATE__"));
	%>
	int nb_line_<%=cid %> = 0;
	<%
    List<IMetadataTable> metadatas = node.getMetadataList();
  	if(metadatas != null && metadatas.size() > 0){
      	IMetadataTable metadata = metadatas.get(0);
      	if(metadata != null){
		%>
			class DBObjectUtil_<%=cid%> {

			        private java.text.SimpleDateFormat isoDateFormat = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

				public DBObjectUtil_<%=cid%>() { isoDateFormat.setTimeZone(java.util.TimeZone.getTimeZone("UTC")); }

                                private com.mongodb.BasicDBObject toMongo (routines.system.Document doc) {
                                    org.dom4j.Element root = doc.getDocument().getRootElement();
                                    return (com.mongodb.BasicDBObject)valueToMongo(root);
                                }
                      
                                private Object valueToMongo (org.dom4j.Element e) {
				    String text = e.getTextTrim();
                                    if (!"".equals(text)) {
				        String type = e.attributeValue("type");
					if (type != null) {
					    type = type.toLowerCase();
					    if (type.equals("int") || type.equals("integer")) {
					        return Integer.parseInt(text);
					    } else if (type.equals("boolean")) {
					        text = text.toLowerCase();
					        if ("false".equals(text)) {
						  return new Boolean(false);
						} else if ("true".equals(text)) {
						  return new Boolean(true);
						} 
					    } else if (type.equals("date") || type.equals("isodate")) {
					        try {
					            return isoDateFormat.parse(text);
						} catch (ParseException ex) {
						    throw new RuntimeException(ex);
                                                }
					    }
					}
                                        return text;
                                    } else if (e.element("item") != null) {
                                        com.mongodb.BasicDBList result = new com.mongodb.BasicDBList();
                                        for (java.util.Iterator<org.dom4j.Element> i = e.elementIterator("item"); i.hasNext();) {
                                            org.dom4j.Element sub = i.next();
                                            result.add(valueToMongo(sub));
                                        }
                                        return result;
                                    } else { 
                                        com.mongodb.BasicDBObject result = new com.mongodb.BasicDBObject();
                                        for (java.util.Iterator<org.dom4j.Element> i = e.elementIterator(); i.hasNext();) {
                                            org.dom4j.Element sub = i.next();
                                            String name = sub.getName();
                                            result.put(name, valueToMongo(sub));
                                        } 
                                        return result;
                                   }
                                }
			
				private com.mongodb.BasicDBObject object = null;
				//Put value to embedded document
				//If have no embedded document, put the value to root document
				public void put(String parentNode, String curentName, Object value) {
				        if (value instanceof routines.system.Document) {
					        object = toMongo((routines.system.Document)value);
					} else if (parentNode == null || "".equals(parentNode)) {
						object.put(curentName, value);
					} else {
						String objNames[]= parentNode.split("\\.");
						com.mongodb.BasicDBObject lastNode = getParentNode(parentNode, objNames.length-1);
						lastNode.put(curentName, value);
						com.mongodb.BasicDBObject parenttNode = null;
						for (int i = objNames.length - 1; i >=0; i--) {
							parenttNode=getParentNode(parentNode, i-1);
							parenttNode.put(objNames[i], lastNode);
							lastNode=(com.mongodb.BasicDBObject)parenttNode.clone();
						}
						object=lastNode;
					}
				}
				//Get node(embedded document) by path configuration
				public com.mongodb.BasicDBObject getParentNode(String parentNode, int index) {
					com.mongodb.BasicDBObject basicDBObject = object;
					if (parentNode == null || "".equals(parentNode)) {
						return object;
					} else {
						String objNames[] = parentNode.split("\\.");
						for (int i = 0; i <= index; i++) {
							basicDBObject = (com.mongodb.BasicDBObject) basicDBObject
									.get(objNames[i]);
							if (basicDBObject == null) {
								basicDBObject = new com.mongodb.BasicDBObject();
								return basicDBObject;
							}
							if (i == index) {
								break;
							}
						}
						return basicDBObject;
					}
				}
				
				public void putkeyNode(String parentNode, String curentName, Object value){
					if (parentNode == null || "".equals(parentNode) || ".".equals(parentNode)) {
						put(parentNode, curentName, value);
					}else{
						put("", parentNode+"."+curentName, value);
					}
				}
			
				public com.mongodb.BasicDBObject getObject() {
					return this.object;
				}
				
				public void setObject(com.mongodb.BasicDBObject object){
					this.object=object;
				}
			
			}
			DBObjectUtil_<%=cid%> updateObjectUtil_<%=cid%>=new DBObjectUtil_<%=cid%>();
			DBObjectUtil_<%=cid%> queryObjectUtil_<%=cid%>=new DBObjectUtil_<%=cid%>();
			java.util.Map<String, String> pathMap_<%=cid%>=new java.util.HashMap<String, String>();
			com.mongodb.Mongo mongo_<%=cid%>=null;
			com.mongodb.DB db_<%=cid%> =null;
			<%
			for (int i=0;i<mapping.size();i++) {
				String path = mapping.get(i).get("PARENT_NODE_PATH");
				if(path==null || "".equals(path)){
					path="\"\"";
				}
				String schemaColumn=mapping.get(i).get("SCHEMA_COLUMN");

				%>
				pathMap_<%=cid%>.put("<%=schemaColumn%>",<%=path%>);
				<%
			}
      		if (useExistingConnection){
      			String connection = ElementParameterParser.getValue(node, "__CONNECTION__");
				%>
				mongo_<%=cid%>=(com.mongodb.Mongo)globalMap.get("mongo_<%=connection%>");
				db_<%=cid%> = (com.mongodb.DB) globalMap.get("db_<%=connection%>");
				<%
				if(isLog4jEnabled){
				%>	
					log.info("<%=cid%> - Get an existing client from " + "<%=connection%>" + ".");
					log.info("<%=cid%> - Get an existing DB from " + "<%=connection%>" + ".");
				<%
				}
			}else{
	 			List<Map<String,String>> replicaAddrs= (List<Map<String,String>>)ElementParameterParser.getObjectValue(node, "__REPLICA_SET__");
	 		    boolean useSSL = "true".equalsIgnoreCase(ElementParameterParser.getValue(node, "__USE_SSL__"));

	 		    String mongoDbDriver = "com.mongodb.MongoClient";

	 		    %>
	 		    // Empty client options
                com.mongodb.MongoClientOptions clientOptions_<%=cid%> = new com.mongodb.MongoClientOptions.Builder().build();
                <%
                if (useSSL) {
                    %>
                    clientOptions_<%=cid%> = new com.mongodb.MongoClientOptions.Builder()
                    .socketFactory(javax.net.ssl.SSLSocketFactory.getDefault())
                    .build();
                    <%
                }
                
                // Client Credentials
                %>
                // Empty client credentials list
                java.util.List<com.mongodb.MongoCredential> mongoCredentialList_<%=cid%> = new java.util.ArrayList<com.mongodb.MongoCredential>();

                <%
                // Authentication
                if (authentication){
                    %>
                    com.mongodb.MongoCredential mongoCredential_<%=cid%>;
                    <%
                    if((authenticationMechanism.equals("NEGOTIATE_MEC"))||(authenticationMechanism.equals("PLAIN_MEC"))||(authenticationMechanism.equals("SCRAMSHA1_MEC"))){
                        String passwordFieldName = "__PASSWORD__";
                        %>
                        <%@ include file="@{org.talend.designer.components.localprovider}/components/templates/password.javajet"%>
                        <%
                        if(authenticationMechanism.equals("NEGOTIATE_MEC")){
                			if(dbversion.equals("MONGODB_3_0_X")) {
%>
                    			mongoCredential_<%=cid%> = com.mongodb.MongoCredential.createCredential(<%=dbuser%>, <%=dbname%>, new String(decryptedPassword_<%=cid%>).toCharArray());
<%                    
                			} else {
%>
                    			mongoCredential_<%=cid%> = com.mongodb.MongoCredential.createMongoCRCredential(<%=dbuser%>, <%=dbname%>, new String(decryptedPassword_<%=cid%>).toCharArray());
<%
                			}
                        } else if(authenticationMechanism.equals("PLAIN_MEC")){
                            %>
                            mongoCredential_<%=cid%> = com.mongodb.MongoCredential.createPlainCredential(<%=dbuser%>, "$external", new String(decryptedPassword_<%=cid%>).toCharArray());
                            <%
                        } else if(authenticationMechanism.equals("SCRAMSHA1_MEC")){
                            %>
                            mongoCredential_<%=cid%> = com.mongodb.MongoCredential.createScramSha1Credential(<%=dbuser%>, <%=dbname%>, new String(decryptedPassword_<%=cid%>).toCharArray());
                            <%
                        }
                    } else { // GSSAPI SASL (KERBEROS)
                        %>
                        System.setProperty("java.security.krb5.realm", <%=krbRealm%>);
                        System.setProperty("java.security.krb5.kdc", <%=krbKdc%>);
                        System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
                        mongoCredential_<%=cid%> = com.mongodb.MongoCredential.createGSSAPICredential(<%=krbUserPrincipal%>);
                        <%
                    }
                    %>
                    mongoCredentialList_<%=cid%>.add(mongoCredential_<%=cid%>);
                    <%
                }

                // the client
                if(useReplicaSet){
                    %>
                    List<com.mongodb.ServerAddress> addrs_<%=cid%> = new java.util.ArrayList<com.mongodb.ServerAddress>();
                    <%
                    for(Map<String,String> replicaAddr:replicaAddrs){
                    %>
                        addrs_<%=cid%>.add(new com.mongodb.ServerAddress(<%=replicaAddr.get("REPLICA_HOST")%>,<%=replicaAddr.get("REPLICA_PORT")%>));
                    <%
                    }
                    %>
                    mongo_<%=cid%> = new <%=mongoDbDriver%>(addrs_<%=cid%>, mongoCredentialList_<%=cid%>, clientOptions_<%=cid%>);
                    <%
                }else{
                    %>
                    com.mongodb.ServerAddress serverAddress_<%=cid%> = new com.mongodb.ServerAddress(<%=dbhost%>, <%=dbport%>);
                    mongo_<%=cid%> = new <%=mongoDbDriver%>(serverAddress_<%=cid%>, mongoCredentialList_<%=cid%>, clientOptions_<%=cid%>);
                    <%
                }
                %>
				resourceMap.put("mongo_<%=cid%>", mongo_<%=cid%>);
				db_<%=cid%> = mongo_<%=cid%>.getDB(<%=dbname%>);
				<%

			}

			if(setWriteConcern){
				%>
				mongo_<%=cid%>.setWriteConcern(com.mongodb.WriteConcern.<%=writeConcern%>);
				<%
			}

			%>
			<%
			if(isLog4jEnabled){
			%>	
				log.info("<%=cid%> - Connecting to " + mongo_<%=cid%>.getServerAddressList() + ".");
			<%
			}
			if(dropExistCollection){
			%>
				if(db_<%=cid%>.collectionExists(<%=collection%>)){
					db_<%=cid%>.getCollection(<%=collection%>).drop();
				}
			<%
			}
			%>
			com.mongodb.DBCollection coll_<%=cid%> = db_<%=cid%>.getCollection(<%=collection%>);
			<%

			// BulkWrite
			if(bulkWrite){
			%>
				com.mongodb.BulkWriteOperation bulkWriteOperation_<%=cid%> = coll_<%=cid%>.initialize<%=bulkWriteType%>BulkOperation();
				int bulkWriteOperationCounter_<%=cid%> = 1;
				int bulkWriteOperationSize_<%=cid%> = Integer.parseInt(<%=bulkWriteOperationSize%>);
				<%
			}

			if (isLog4jEnabled) {
			%>
				log.info("<%=cid%> - Start to write data into database [" + db_<%=cid%>.getName() + "].");
			<%
			}
    	}
    }
	%>