import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
public static ResultSequence runQuery(String xccUri, String query, String queryLanguage) 
        throws XccConfigException, URISyntaxException, RequestException {
    ContentSource cs = csMap.get(xccUri);
    if (cs == null) {
        cs = ContentSourceFactory.newContentSource(new URI(
        xccUri));
        csMap.put(xccUri, cs);
    }
    session = cs.newSession();
    AdhocQuery aquery = session.newAdhocQuery(query);

    RequestOptions options = new RequestOptions();
    options.setQueryLanguage(queryLanguage);
    options.setCacheResult(false);
    options.setDefaultXQueryVersion("1.0-ml");
    aquery.setOptions(options);
    return session.submitRequest(aquery);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
public static ResultSequence runQueryAgainstDb(String xccUri, String query, String db)
    throws XccConfigException, URISyntaxException, RequestException {
    StringBuilder buf = new StringBuilder();
    buf.append("xdmp:eval('");
    buf.append(query);
    buf.append("',(),<options xmlns=\"xdmp:eval\">\n" + 
            "            <database>{xdmp:database(\"");
    buf.append(db);
    buf.append("\")}</database>\n" + 
            "          </options>)");
    ContentSource cs = csMap.get(xccUri);
    if (cs == null) {
        cs = ContentSourceFactory.newContentSource(new URI(
        xccUri));
        csMap.put(xccUri, cs);
    }
    session = cs.newSession();
    AdhocQuery aquery = session.newAdhocQuery(buf.toString());

    RequestOptions options = new RequestOptions();
    options.setCacheResult(false);
    options.setDefaultXQueryVersion("1.0-ml");
    aquery.setOptions(options);
    return session.submitRequest(aquery);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
public void checkOutputSpecs(Configuration conf, ContentSource cs) 
throws IOException {
    // check for required configuration
    if (conf.get(OUTPUT_QUERY) == null) {
        throw new IllegalArgumentException(OUTPUT_QUERY + 
        " is not specified.");
    }
    // warn against unsupported configuration
    if (conf.get(BATCH_SIZE) != null) {
        LOG.warn("Config entry for " +
                "\"mapreduce.marklogic.output.batchsize\" is not " +
                "supported for " + this.getClass().getName() + 
                " and will be ignored.");
    }
    String queryLanguage = conf.get(OUTPUT_QUERY_LANGUAGE);
    if (queryLanguage != null) {
        InternalUtilities.checkQueryLanguage(queryLanguage);
    }
    // store hosts into config system
    DefaultStringifier.store(conf, queryHosts(cs), OUTPUT_FOREST_HOST);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
/**
 * Get the session for this writer.  One writer only maintains one session.
 * 
 * @return Session for this writer.
 * @throws IOException
 */
protected Session getSession() throws IOException {
    if (session == null) {
        // start a session
        try {
            ContentSource cs = InternalUtilities.getOutputContentSource(
                    conf, hostName);
            session = cs.newSession();
            if (LOG.isDebugEnabled()) {
                LOG.debug("Connect to " + session.getConnectionUri().getHost());
            }
            if (txnSize > 1) {
                session.setTransactionMode(TransactionMode.UPDATE);
            }
        } catch (XccConfigException e) {
            LOG.error("Error creating a new session: ", e);
            throw new IOException(e);
        }    
    }
    return session;
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
/**
 * Get input content source.
 *
 * @param conf job configuration
 * @param host host to connect to
 * @return content source
 * @throws IOException 
 * @throws XccConfigException
 */
public static ContentSource getInputContentSource(Configuration conf,
        String host) 
throws XccConfigException, IOException {      
    String user = conf.get(INPUT_USERNAME, "");
    String password = conf.get(INPUT_PASSWORD, "");
    String port = conf.get(INPUT_PORT,"8000");
    String db = conf.get(INPUT_DATABASE_NAME);
    int portInt = Integer.parseInt(port);
    boolean useSsl = conf.getBoolean(INPUT_USE_SSL, false);
    if (useSsl) {
        return getSecureContentSource(host, portInt, user, password, 
                    db, getInputSslOptions(conf));
    }
    return ContentSourceFactory.newContentSource(host, portInt, 
            user, password, db);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
static ContentSource getSecureContentSource(String host, int port,
        String user, String password, String db, 
        SslConfigOptions sslOptions) 
throws XccConfigException {
    ContentSource contentSource = null;
  
    // construct XCC SecurityOptions
    SecurityOptions options;
    try {
        options = new SecurityOptions(sslOptions.getSslContext());
    } catch (KeyManagementException | NoSuchAlgorithmException e) {
        throw new XccConfigException("Error constructing SecurityOptions",
                e);
    }
    options.setEnabledCipherSuites(sslOptions.getEnabledCipherSuites());
    options.setEnabledProtocols(sslOptions.getEnabledProtocols());
  
    // construct content source
    contentSource = ContentSourceFactory.newContentSource(
            host, port, user, password, db, options);        
 
    return contentSource;
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
/**
 * Get output content source.
 *
 * @param conf job configuration
 * @param hostName host name
 * @return content source
 * @throws IOException 
 * @throws XccConfigException 
 * @throws IOException 
 */
public static ContentSource getOutputContentSource(Configuration conf,
        String hostName) 
throws XccConfigException, IOException {
    String user = conf.get(OUTPUT_USERNAME, "");
    String password = conf.get(OUTPUT_PASSWORD, "");
    String port = conf.get(OUTPUT_PORT,"8000");
    String db = conf.get(OUTPUT_DATABASE_NAME);
    int portInt = Integer.parseInt(port);
    boolean useSsl = conf.getBoolean(OUTPUT_USE_SSL, false);
    if (useSsl) {
        return getSecureContentSource(hostName, portInt, user, password,
                    db, getOutputSslOptions(conf));
    }
    return ContentSourceFactory.newContentSource(hostName, portInt, 
            user, password, db);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
protected Runnable buildRunnable(final ContentSource contentSource, final List<? extends DocumentWriteOperation> items) {
	return new Runnable() {
		@Override
		public void run() {
			Session session = contentSource.newSession();
			int count = items.size();
			Content[] array = new Content[count];
			for (int i = 0; i < count; i++) {
				array[i] = documentWriteOperationAdapter.adapt(items.get(i));
			}
			if (logger.isDebugEnabled()) {
				logger.debug("Writing " + count + " documents to MarkLogic");
			}
			try {
				session.insertContent(array);
				if (logger.isInfoEnabled()) {
					logger.info("Wrote " + count + " documents to MarkLogic");
				}
			} catch (RequestException e) {
				throw new RuntimeException("Unable to insert content: " + e.getMessage(), e);
			} finally {
				session.close();
			}
		}
	};
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
protected int execCommand(String command, Map<String, Parameter> params) throws RequestException {
       if ("insertDocument".equals(command)) {

       	ContentCreateOptions options = ContentCreateOptions.newXmlInstance();
       	//options.setNamespace("MyNameSpace");
           String collect = params.get("collect").getName();
           if (collect != null) {
           	options.setCollections(new String[] {collect});
           }
           String prefix = params.get("prefix").getName();
           if (prefix == null) {
           	prefix = "doc";
           }
           String uri = prefix + cnt.getAndIncrement() + ".xml";
           String doc = params.get("doc").getName();
       	Content content = ContentFactory.newContent(uri, doc, options);
   		ContentSource xcs = getContentSource();
   		Session xss = xcs.newSession();
       	xss.insertContent(content);       
       	xss.close();
           return 1;
       }
	return 0;
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
protected int execQuery(String query, Map<String, Parameter> params) throws RequestException {
	
	ContentSource xcs = getContentSource();
	Session xss = xcs.newSession();
	//xss.getDefaultRequestOptions().setResultBufferSize(fetchSize);
	Request request = xss.newAdhocQuery(query);
	bindParams(params, request);
	ResultSequence rs = xss.submitRequest(request);		
    int cnt = 0;
    if (fetchSize > 0) {
    	while (rs.hasNext() && cnt < fetchSize) {
    		rs.next();
    		cnt++;
    	}
    } else {
    	while (rs.hasNext()) {
    		rs.next();
    		cnt++;
    	}
    }
    rs.close();
    xss.close();
    return cnt;
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
public void runApp() throws Exception {
    URI uri = new URI(String.format(XCC_CONNECTION_URI_TEMPLATE, config.getUsername(), config.getPassword(), config.getHost(), config.getPort(), config.getDatabaseName()));
    ContentSource contentSource = ContentSourceFactory.newContentSource(uri);
    try (Session session = contentSource.newSession()) {
        Request request = session.newAdhocQuery(readFile(config.getMainFile()));
        for (XQueryRunnerVariable variable : config.getVariables()) {
            if (variable.ACTIVE) {
                request.setNewVariable(getName(variable.NAME, variable.NAMESPACE), getXdmValue(variable.TYPE,
                        variable.VALUE));
            }
        }
        ResultSequence rs = session.submitRequest(request);
        output.println(rs.asString());
    }
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
public RecordWriter<DocumentURI, VALUEOUT> getRecordWriter(
    TaskAttemptContext context) throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();
    fastLoad = Boolean.valueOf(conf.get(OUTPUT_FAST_LOAD));
    Map<String, ContentSource> sourceMap = getSourceMap(fastLoad, context);
    getMimetypesMap();
    // construct the ContentWriter
    return new TransformWriter<VALUEOUT>(conf, sourceMap, fastLoad, am);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
public void checkOutputSpecs(Configuration conf, ContentSource cs)
    throws IOException {
    super.checkOutputSpecs(conf, cs);

    // store mimetypes map into config system
    DefaultStringifier.store(conf, getMimetypesMap(),
        ConfigConstants.CONF_MIMETYPES);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
public DatabaseContentWriter(Configuration conf,
        Map<String, ContentSource> hostSourceMap, boolean fastLoad,
        AssignmentManager am) {
    super(conf, hostSourceMap, fastLoad, am);

    if (countBased) {
        metadatas = new URIMetadata[1][batchSize];
    } else {
        metadatas = new URIMetadata[forestIds.length][batchSize];
    }
    isCopyProps = conf.getBoolean(CONF_COPY_PROPERTIES, true);
    isCopyPerms = conf.getBoolean(CONF_COPY_PERMISSIONS, true);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
public RecordWriter<DocumentURI, DatabaseDocumentWithMeta> getRecordWriter(
    TaskAttemptContext context) throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();
    fastLoad = Boolean.valueOf(conf.get(OUTPUT_FAST_LOAD));
    Map<String, ContentSource> sourceMap = getSourceMap(fastLoad, context);
    // construct the DatabaseTransformContentWriter
    return new DatabaseTransformWriter<DatabaseDocumentWithMeta>(conf,
        sourceMap, fastLoad, am);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
public RecordWriter<DocumentURI, DatabaseDocumentWithMeta> getRecordWriter(
    TaskAttemptContext context) throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();
    fastLoad = Boolean.valueOf(conf.get(OUTPUT_FAST_LOAD));
    Map<String, ContentSource> sourceMap = getSourceMap(fastLoad, context);
    // construct the DatabaseContentWriter
    return new DatabaseContentWriter<DatabaseDocumentWithMeta>(conf,
        sourceMap, fastLoad, am);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
public DatabaseTransformWriter(Configuration conf,
        Map<String, ContentSource> hostSourceMap, boolean fastLoad,
        AssignmentManager am) {
    super(conf, hostSourceMap, fastLoad, am);

    isCopyProps = conf.getBoolean(CONF_COPY_PROPERTIES, true);
    isCopyPerms = conf.getBoolean(CONF_COPY_PERMISSIONS, true);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
public static ResultSequence assertDocsFormat(String xccUri, String docType) 
        throws XccConfigException, URISyntaxException, RequestException {
    ContentSource cs = csMap.get(xccUri);
    if (cs == null) {
        cs = ContentSourceFactory.newContentSource(new URI(
        xccUri));
        csMap.put(xccUri, cs);
    }
    
    session = cs.newSession();
    String query = "function testDocument() {"
            + "var it=fn.doc();"
            + "for (var u of it) {"
            + "if (u.documentFormat != '" + docType +"') {"
            + "return false;"
            + "}}"
            + "return true;}"
            + "testDocument();";
    AdhocQuery aquery = session.newAdhocQuery(query);
    
    RequestOptions options = new RequestOptions();
    options.setCacheResult(false);
    options.setQueryLanguage("javascript");
    aquery.setOptions(options);
    
    return session.submitRequest(aquery);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
public void checkOutputSpecs(Configuration conf, ContentSource cs) 
throws IOException {
    // warn against unsupported configuration
    if (conf.get(BATCH_SIZE) != null) {
        LOG.warn("Config entry for " +
                "\"mapreduce.marklogic.output.batchsize\" is not " +
                "supported for " + this.getClass().getName() + 
                " and will be ignored.");
    }     
    // store hosts into config system
    DefaultStringifier.store(conf, queryHosts(cs), OUTPUT_FOREST_HOST);
}
 

import com.marklogic.xcc.ContentSource; //导入依赖的package包/类
@Override
public RecordWriter<DocumentURI, VALUEOUT> getRecordWriter(
        TaskAttemptContext context) throws IOException, InterruptedException {
    Configuration conf = context.getConfiguration();
    // TODO: if MAPREDUCE-3377 still exists, need to re-run initialize
    fastLoad = Boolean.valueOf(conf.get(OUTPUT_FAST_LOAD));
    Map<String, ContentSource> sourceMap = getSourceMap(fastLoad, context);
    // construct the ContentWriter
    return new ContentWriter<VALUEOUT>(conf, sourceMap, fastLoad, 
            am);
}