text:
package info.rmapproject.core.rmapservice.impl.rdf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.model.Resource;
import org.eclipse.rdf4j.model.Statement;
import org.eclipse.rdf4j.model.Value;
import org.eclipse.rdf4j.model.vocabulary.RDF;
import org.eclipse.rdf4j.query.Binding;
import org.eclipse.rdf4j.query.BindingSet;

import info.rmapproject.core.exception.RMapDefectiveArgumentException;
import info.rmapproject.core.exception.RMapException;
import info.rmapproject.core.model.impl.rdf4j.ORAdapter;
import info.rmapproject.core.model.request.OrderBy;
import info.rmapproject.core.model.request.RMapSearchParams;
import info.rmapproject.core.rmapservice.impl.rdf4j.triplestore.Rdf4jSparqlUtils;
import info.rmapproject.core.rmapservice.impl.rdf4j.triplestore.Rdf4jTriplestore;
import info.rmapproject.core.vocabulary.impl.rdf4j.PROV;
import info.rmapproject.core.vocabulary.impl.rdf4j.RMAP;

/**
 *  A concrete class for managing RMap Resources, implemented using RDF4J
 *
 * @author khanson
 * @author smorrissey
 */

public class ORMapResourceMgr extends ORMapObjectMgr {

	/**
	 * Get list of DiSCO IRIs that have a statement containing the resource.
	 *
	 * @param resource resource a Resource IRI
	 * @param params the search filter parameters
	 * @param ts the triplestore instance
	 * @return a set of IRIs of DiSCOs that contain a statement referencing the Resource
	 * @throws RMapException the RMap exception
	 * @throws RMapDefectiveArgumentException the RMap defective argument exception
	 */
	public List<IRI> getResourceRelatedDiSCOS(IRI resource, RMapSearchParams params, Rdf4jTriplestore ts)
						throws RMapException, RMapDefectiveArgumentException {

	//query gets discoIds and startDates of created DiSCOs that contain Resource
	/*  SELECT DISTINCT ?rmapObjId ?startDate
		WHERE {
		GRAPH ?rmapObjId
		  {
		     {?s ?p <http://dx.doi.org/10.1109/InPar.2012.6339604>} UNION
		        {<http://dx.doi.org/10.1109/InPar.2012.6339604> ?p ?o} .
		     ?discoId <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://purl.org/ontology/rmap#DiSCO> .
		  } .
		GRAPH ?eventId {
		 	?eventId <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://purl.org/ontology/rmap#Event> .
			{?eventId <http://www.w3.org/ns/prov#generated> ?rmapObjId} UNION
			{?eventId <http://purl.org/ontology/rmap#derivedObject> ?rmapObjId} .
		 	?eventId <http://www.w3.org/ns/prov#startedAtTime> ?startDate .
		 	{?eventId <http://www.w3.org/ns/prov#wasAssociatedWith> <ark:/22573/rmd18nd2m3>} UNION
		 	{?eventId <http://www.w3.org/ns/prov#wasAssociatedWith> <ark:/22573/rmd18nd2p4>} .
		 	}
		  }
		*/

		List<IRI> discos = getRelatedObjects(resource, params, ts, RMAP.DISCO);
		return discos;
	}

	/**
	 * Get list of RMap Agent IRIs that have a statement containing the resource.
	 *
	 * @param resource a Resource IRI
	 * @param ts the triplestore instance
	 * @return a set of IRIs for Agents that asserted a statement about the Resource
	 * @throws RMapException the RMap exception
	 * @throws RMapDefectiveArgumentException the RMap defective argument exception
	 */
	public List<IRI> getResourceAssertingAgents(IRI resource, RMapSearchParams params, Rdf4jTriplestore ts)
					throws RMapException, RMapDefectiveArgumentException {

		if (resource==null){
			throw new RMapDefectiveArgumentException ("Null value provided for the Resource IRI");
		}

		Set<org.eclipse.rdf4j.model.IRI> systemAgents = ORAdapter.uriSet2Rdf4jIriSet(params.getSystemAgents());

		String sResource = Rdf4jSparqlUtils.convertIriToSparqlParam(resource);
		String sysAgentSparql = Rdf4jSparqlUtils.convertSysAgentIriListToSparqlFilter(systemAgents);
		String statusFilterSparql = Rdf4jSparqlUtils.convertRMapStatusToSparqlFilter(params.getStatusCode(), "?rmapObjId");
		String dateFilterSparql = Rdf4jSparqlUtils.convertDateRangeToSparqlFilter(params.getDateRange(), "?startDate");
		String limitOffsetSparql = Rdf4jSparqlUtils.convertLimitOffsetToSparqlFilter(params.getLimitForQuery(), params.getOffset());

		StringBuilder sparqlQuery =
				new StringBuilder("SELECT DISTINCT ?agentId "
							+ "WHERE { "
							+ " GRAPH ?rmapObjId "
							+ "	  {"
							+ "		{?s ?p " + sResource + "} UNION "
							+ "		  {" + sResource + " ?p ?o} ."
							+ "	  } . "
							+ " GRAPH ?eventId {"
							+ "   ?eventId <" + RDF.TYPE + "> <" + RMAP.EVENT + "> ."
							+ "   {?eventId <" + PROV.GENERATED + "> ?rmapObjId} UNION"
							+ "   {?eventId <" + RMAP.DERIVEDOBJECT + "> ?rmapObjId} ."
							+ "	  ?eventId <" + PROV.WASASSOCIATEDWITH + "> ?agentId . ");

		if (dateFilterSparql.length()>0){
			sparqlQuery.append("   ?eventId <" + PROV.STARTEDATTIME + "> ?startDate .");
		}
		sparqlQuery.append(sysAgentSparql
							+ "  } "
							+ statusFilterSparql
							+ dateFilterSparql
							+ "} ");
		if (params.getOrderBy()==OrderBy.SELECT_ORDER){
			sparqlQuery.append("ORDER BY ?agentId");
		}
		sparqlQuery.append(limitOffsetSparql);

		List<IRI> assertingAgents = Rdf4jSparqlUtils.bindQueryToIriList(sparqlQuery.toString(), ts, "agentId");

		return assertingAgents;

	}

	/**
	 * Get set of RMap object IRIs that have a statement containing the resource, filtered by type (DiSCO or Agent).
	 *
	 * @param resource a Resource IRI
	 * @param the search filter parameters
	 * @param ts the triplestore instance
	 * @param rmapType the RMap type to filter by
	 * @return Set of IRIs for RMap objects that contain a reference to the Resource ID
	 * @throws RMapException the RMap exception
	 * @throws RMapDefectiveArgumentException the RMap defective argument exception
	 */
	protected List<IRI> getRelatedObjects(IRI resource, RMapSearchParams params, Rdf4jTriplestore ts, IRI rmapType)
					throws RMapException, RMapDefectiveArgumentException {
		if (resource==null){
			throw new RMapDefectiveArgumentException ("Null value provided for the Resource IRI");
		}

		Set<org.eclipse.rdf4j.model.IRI> systemAgents = ORAdapter.uriSet2Rdf4jIriSet(params.getSystemAgents());

		String sResource = Rdf4jSparqlUtils.convertIriToSparqlParam(resource);
		String sysAgentSparql = Rdf4jSparqlUtils.convertSysAgentIriListToSparqlFilter(systemAgents);
		String statusFilterSparql = Rdf4jSparqlUtils.convertRMapStatusToSparqlFilter(params.getStatusCode(), "?rmapObjId");
		String dateFilterSparql = Rdf4jSparqlUtils.convertDateRangeToSparqlFilter(params.getDateRange(), "?startDate");
		String limitOffsetSparql = Rdf4jSparqlUtils.convertLimitOffsetToSparqlFilter(params.getLimitForQuery(), params.getOffset());

		StringBuilder sparqlQuery =
				new StringBuilder("SELECT DISTINCT ?rmapObjId "
							+ "WHERE { "
							+ " GRAPH ?rmapObjId "
							+ "	  {"
							+ "		{?s ?p " + sResource + "} UNION "
							+ "		  {" + sResource + " ?p ?o} ."
							+ "     ?rmapObjId <" + RDF.TYPE + "> <" + rmapType + "> . "
							+ "	  } . "
							+ " GRAPH ?eventId {"
							+ "   ?eventId <" + RDF.TYPE + "> <" + RMAP.EVENT + "> ."
							+ "   {?eventId <" + PROV.GENERATED + "> ?rmapObjId} UNION"
							+ "   {?eventId <" + RMAP.DERIVEDOBJECT + "> ?rmapObjId} .");

		if (dateFilterSparql.length()>0){
			sparqlQuery.append("   ?eventId <" + PROV.STARTEDATTIME + "> ?startDate .");
		}
		sparqlQuery.append(sysAgentSparql
							+ "  } "
							+ statusFilterSparql
							+ dateFilterSparql
							+ "} ");
		if (params.getOrderBy()==OrderBy.SELECT_ORDER){
			sparqlQuery.append("ORDER BY ?rmapObjId");
		}
		sparqlQuery.append(limitOffsetSparql);

		List<IRI> rmapObjectIds = Rdf4jSparqlUtils.bindQueryToIriList(sparqlQuery.toString(), ts, "rmapObjId");

		return rmapObjectIds;

	}

	/**
	 * Get list of Events IRIs that are associated with an object that references the resource passed in.
	 *
	 * @param resource a Resource IRI
	 * @param the search filter parameters
	 * @param ts the triplestore instance
	 * @return a set of IRIs for Events that produced a statement referencing the Resource IRI
	 * @throws RMapException the RMap exception
	 * @throws RMapDefectiveArgumentException the RMap defective argument exception
	 */
	public List<IRI> getResourceRelatedEvents(IRI resource, RMapSearchParams params, Rdf4jTriplestore ts)
						throws RMapException, RMapDefectiveArgumentException {
		if (resource==null){
			throw new RMapDefectiveArgumentException ("null IRI");
		}
		String sResource = Rdf4jSparqlUtils.convertIriToSparqlParam(resource);

		Set<org.eclipse.rdf4j.model.IRI> systemAgents = ORAdapter.uriSet2Rdf4jIriSet(params.getSystemAgents());
		String sysAgentSparql = Rdf4jSparqlUtils.convertSysAgentIriListToSparqlFilter(systemAgents);
		String dateFilterSparql = Rdf4jSparqlUtils.convertDateRangeToSparqlFilter(params.getDateRange(), "?startDate");
		String limitOffsetSparql = Rdf4jSparqlUtils.convertLimitOffsetToSparqlFilter(params.getLimitForQuery(), params.getOffset());
		String statusFilterSparql = Rdf4jSparqlUtils.convertRMapStatusToSparqlFilter(params.getStatusCode(), "?rmapObjId");

		//query gets eventIds and startDates of created DiSCOs that contain Resource
		/*  SELECT DISTINCT ?eventId ?startDate
			WHERE {
			GRAPH ?rmapObjId
			  {
			     {?s ?p <http://dx.doi.org/10.1109/InPar.2012.6339604>} UNION
			        {<http://dx.doi.org/10.1109/InPar.2012.6339604> ?p ?o} .
			  } .
			GRAPH ?eventId {
			 	?eventId <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://purl.org/ontology/rmap#Event> .
				?eventId ?eventtype ?rmapObjId .
			 	?eventId <http://www.w3.org/ns/prov#startedAtTime> ?startDate .
			 	{?eventId <http://www.w3.org/ns/prov#wasAssociatedWith> <ark:/22573/rmd18nd2m3>} UNION
			 	{?eventId <http://www.w3.org/ns/prov#wasAssociatedWith> <ark:/22573/rmd18nd2p4>} .
			 	}
			}
			ORDER BY ?startDate ?eventId
			LIMIT 10 OFFSET 10
			*/
		StringBuilder sparqlQuery =
				new StringBuilder("SELECT DISTINCT ?eventId "
							+ "WHERE { "
							+ " GRAPH ?rmapObjId "
							+ "	  {"
							+ "		{?s ?p " + sResource + "} UNION "
							+ "		  {" + sResource + " ?p ?o} ."
							+ "	  } . "
							+ " GRAPH ?eventId {"
							+ "   ?eventId <" + RDF.TYPE + "> <" + RMAP.EVENT + "> ."
							+ "   {?eventId ?eventType ?rmapObjId} .");

		if (dateFilterSparql.length()>0){
			sparqlQuery.append("   ?eventId <" + PROV.STARTEDATTIME + "> ?startDate .");
		}
		sparqlQuery.append(sysAgentSparql
							+ "  } "
							+ statusFilterSparql
							+ dateFilterSparql
							+ "} ");
		if (params.getOrderBy()==OrderBy.SELECT_ORDER){
			sparqlQuery.append("ORDER BY ?eventId");
		}
		sparqlQuery.append(limitOffsetSparql);

		List<IRI> events = Rdf4jSparqlUtils.bindQueryToIriList(sparqlQuery.toString(), ts, "eventId");
		return events;
	}

	/**
	 * Get Statements referencing a IRI in subject or object, whose Subject, Predicate, and Object comprise an RMapStatement,
	 * and (if statusCode is not null), whose status matches statusCode, agent, and date filters.
	 *
	 * @param resource a Resource IRI
	 * @param the search filter parameters
	 * @param ts the triplestore instance
	 * @return set of Statements that reference the Resource IRI as either the subject or object
	 * @throws RMapDefectiveArgumentException the RMap defective argument exception
	 * @throws RMapException the RMap exception
	 */
	public List<Statement> getRelatedTriples(IRI resource, RMapSearchParams params, Rdf4jTriplestore ts) {
		return getRelatedTriples(resource, null, params, ts);
	}

	/**
	 * Get Statements referencing a IRI in subject or object, whose Subject, Predicate, and Object comprise an RMapStatement,
	 * and (if statusCode is not null), whose status matches statusCode, agent, and date filters.
	 *
	 * @param resource a Resource IRI
	 * @param context a context filter (e.g. when retrieving from a single DiSCO use a DiSCO IRI)
	 * @param the search filter parameters
	 * @param ts the triplestore instance
	 * @return set of Statements that reference the Resource IRI as either the subject or object
	 * @throws RMapDefectiveArgumentException the RMap defective argument exception
	 * @throws RMapException the RMap exception
	 */
	public List<Statement> getRelatedTriples(IRI resource, IRI context, RMapSearchParams params, Rdf4jTriplestore ts)
			throws RMapDefectiveArgumentException, RMapException {
		if (resource==null){
			throw new RMapDefectiveArgumentException ("null URI");
		}
		List<Statement> relatedStmts = new ArrayList<Statement>();

		Set<IRI> systemAgents = ORAdapter.uriSet2Rdf4jIriSet(params.getSystemAgents());
		String sResource = Rdf4jSparqlUtils.convertIriToSparqlParam(resource);
		String sysAgentSparql = Rdf4jSparqlUtils.convertSysAgentIriListToSparqlFilter(systemAgents);
		String statusFilterSparql = Rdf4jSparqlUtils.convertRMapStatusToSparqlFilter(params.getStatusCode(), "?rmapObjId");
		String dateFilterSparql = Rdf4jSparqlUtils.convertDateRangeToSparqlFilter(params.getDateRange(), "?startDate");
		String limitOffsetFilterSparql = Rdf4jSparqlUtils.convertLimitOffsetToSparqlFilter(params.getLimitForQuery(), params.getOffset());
		String objectTypeFilterSparql = Rdf4jSparqlUtils.convertObjectExclusionsToFilter(params.excludeIRIs(), params.excludeLiterals());
		String excludeTypesFilterSparql = Rdf4jSparqlUtils.convertTypeExclusionToFilter(params.excludeTypes());

		//query gets eventIds and startDates of created DiSCOs that contain Resource
		/*  SELECT DISTINCT ?s ?p ?o ?startDate
			WHERE {
			GRAPH ?rmapObjId
			 {
				 {
                 ?s ?p ?o
          		 FILTER (?s=<http://dx.doi.org/10.1145/357456.357463>)
       			 }	UNION
				 {
                 ?s ?p ?o
          		 FILTER (?o=<http://dx.doi.org/10.1145/357456.357463>)
       			 }
			  } .
			GRAPH ?eventId {
			 	?eventId <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://purl.org/ontology/rmap#Event> .
				{?eventId <http://www.w3.org/ns/prov#generated> ?rmapObjId} UNION
				{?eventId <http://purl.org/ontology/rmap#derivedObject> ?rmapObjId} .
			 	?eventId <http://www.w3.org/ns/prov#startedAtTime> ?startDate .
			 	{?eventId <http://www.w3.org/ns/prov#wasAssociatedWith> <ark:/22573/rmd18nd2m3>} UNION
			 	{?eventId <http://www.w3.org/ns/prov#wasAssociatedWith> <ark:/22573/rmd18nd2p4>} .
			 	} .
				FILTER NOT EXISTS {?statusChangeEventId <http://purl.org/ontology/rmap#tombstonedObject> ?rmapObjId} .
				FILTER NOT EXISTS {?statusChangeEventId <http://purl.org/ontology/rmap#inactivatedObject> ?rmapObjId} .
				FILTER (?startDate >= "2016-03-22T10:20:13"^^xsd:dateTime) .
                FILTER (?startDate <= "2016-03-23T10:20:13"^^xsd:dateTime)
                                             }
				}
				ORDER BY ?startDate ?s ?p ?o
				LIMIT 10 OFFSET 10

			}
			}
			*/
		StringBuilder sparqlQuery =
				new StringBuilder("SELECT DISTINCT ?s ?p ?o "
							+ "WHERE { "
							+ " GRAPH ?rmapObjId "
							+ "	  {"
							+ "	  	{?s ?p ?o FILTER (?s=" + sResource + ")}	UNION "
							+ "	  	{?s ?p ?o FILTER (?o=" + sResource + ")} . "
							+ "	  } . "
							+ " GRAPH ?eventId {"
							+ "   ?eventId <" + RDF.TYPE + "> <" + RMAP.EVENT + "> ."
							+ "   {?eventId <" + PROV.GENERATED + "> ?rmapObjId} UNION"
							+ "   {?eventId <" + RMAP.DERIVEDOBJECT + "> ?rmapObjId} .");

		if (dateFilterSparql.length()>0){
			sparqlQuery.append("   ?eventId <" + PROV.STARTEDATTIME + "> ?startDate .");
		}
		sparqlQuery.append(sysAgentSparql
							+ "  } "
							+ dateFilterSparql
							+ statusFilterSparql
							+ excludeTypesFilterSparql
							+ objectTypeFilterSparql
							+ "} ");

		if (params.getOrderBy()==OrderBy.SELECT_ORDER){
			sparqlQuery.append("ORDER BY ?s ?p ?o");
		}
		sparqlQuery.append(limitOffsetFilterSparql);

		//if there a context filter was provided, let's be specific about the graph name
		String query = sparqlQuery.toString();
		if (context!=null){
			String graphid = Rdf4jSparqlUtils.convertIriToSparqlParam(context);
			query = query.replaceAll("\\?rmapObjId", graphid);
		}

		List<BindingSet> resultset = null;
		try {
			resultset = ts.getSPARQLQueryResults(query);
		}
		catch (Exception e) {
			throw new RMapException("Could not retrieve SPARQL query results using " + query, e);
		}

		try{
			for (BindingSet bindingSet : resultset){
				Binding subjBinding = bindingSet.getBinding("s");
				Resource subj = (Resource) subjBinding.getValue();
				IRI pred = (IRI) bindingSet.getBinding("p").getValue();
				Binding objBinding = bindingSet.getBinding("o");
				Value obj = objBinding.getValue();
				Statement stmt = ORAdapter.getValueFactory().createStatement(subj, pred, obj);
				relatedStmts.add(stmt);
			}
		}
		catch (RMapException r){throw r;}
		catch (Exception e){
			throw new RMapException("Could not process SPARQL results for resource's related triples", e);
		}

		return relatedStmts;
	}

	/**
	 * Find types of resource in specific context.
	 *
	 * @param rIri IRI of the resource whose type is being checked
	 * @param cIri IRI of the RMap Object in which to check for a type
	 * @param ts the triplestore instance
	 * @return Set of IRIs that are type(s) of resource in given RMap Object, or null if none found
	 * @throws RMapException the RMap exception
	 * @throws RMapDefectiveArgumentException the RMap defective argument exception
	 */
	public List<IRI> getResourceRdfTypes(IRI resourceIri, IRI contextIri,  Rdf4jTriplestore ts)
			throws RMapException, RMapDefectiveArgumentException {
		if (resourceIri==null || contextIri == null || ts == null){
			throw new RMapDefectiveArgumentException ("Null parameter");
		}
		Set<Statement> triples = null;
		try {
		    if (ts.getConnection().size(contextIri)>0) {
		        triples = ts.getStatements(resourceIri, RDF.TYPE, null, contextIri);
		    }
		} catch (Exception e) {
			throw new RMapException (e);
		}
		List<IRI> returnSet = null;
		if (triples!=null && triples.size()>0){
			returnSet = new ArrayList<IRI>();
			for (Statement stmt:triples){
				Value object = stmt.getObject();
				if (object instanceof IRI){
					returnSet.add((IRI)object);
				}
			}
			// correct if only triples found had non-IRI objects
			if (returnSet.size()==0){
				returnSet = null;
			}
		}
		return returnSet;
	}

	/**
	 * Find rdf:types of resource in an RMap Object (Agent or DiSCO)
	 *
	 * @param resourceIri the resource IRI
	 * @param the search filter parameters
	 * @param ts the triplestore instance
	 * @return Map from RMap Object ID to any types found for that resource in that object, or null if no type statement
	 * found for resource in any RMap Object
	 * @throws RMapException the RMap exception
	 * @throws RMapDefectiveArgumentException the RMap defective argument exception
	 */
	public Map<IRI, Set<IRI>> getResourceRdfTypesAllContexts(IRI resourceIri, RMapSearchParams params, Rdf4jTriplestore ts)
			throws RMapException, RMapDefectiveArgumentException {
		if (resourceIri==null || ts == null || params == null){
			throw new RMapDefectiveArgumentException ("Null parameter");
		}

		/*
		PREFIX xsd: <http://www.w3.org/2001/XMLSchema#>
			SELECT DISTINCT ?rmapObjId ?type
						WHERE {
						GRAPH ?rmapObjId
						 {
							 {<http://dx.doi.org/10.1109/ACSSC.1994.471497> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> ?type} .
						  } .
						GRAPH ?eventId {
						 	?eventId <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://purl.org/ontology/rmap#Event> .
							{?eventId <http://www.w3.org/ns/prov#generated> ?rmapObjId} UNION
							{?eventId <http://purl.org/ontology/rmap#derivedObject> ?rmapObjId} .
							?eventId <http://www.w3.org/ns/prov#startedAtTime> ?startDate .
						 	{?eventId <http://www.w3.org/ns/prov#wasAssociatedWith> <ark:/22573/rmd18m7mj4>} . #UNION
						 	{?eventId <http://www.w3.org/ns/prov#wasAssociatedWith> <ark:/22573/rmd18nd2p4>} .
						 	} .
							FILTER NOT EXISTS {?statusChangeEventId <http://purl.org/ontology/rmap#tombstonedObject> ?rmapObjId} .
							FILTER NOT EXISTS {?statusChangeEventId <http://purl.org/ontology/rmap#inactivatedObject> ?rmapObjId} .
							FILTER (?startDate >= "2011-03-22T10:20:13"^^xsd:dateTime) .
			                FILTER (?startDate <= "2016-03-25T10:20:13"^^xsd:dateTime)
			}
			ORDER BY ?rmapObjId ?type
			LIMIT 30 OFFSET 0
		 */

		Set<org.eclipse.rdf4j.model.IRI> systemAgents = ORAdapter.uriSet2Rdf4jIriSet(params.getSystemAgents());
		String sResourceIri = Rdf4jSparqlUtils.convertIriToSparqlParam(resourceIri);
		String sysAgentSparql = Rdf4jSparqlUtils.convertSysAgentIriListToSparqlFilter(systemAgents);
		String statusFilterSparql = Rdf4jSparqlUtils.convertRMapStatusToSparqlFilter(params.getStatusCode(), "?rmapObjId");
		String dateFilterSparql = Rdf4jSparqlUtils.convertDateRangeToSparqlFilter(params.getDateRange(), "?startDate");

		String limitOffsetFilterSparql = Rdf4jSparqlUtils.convertLimitOffsetToSparqlFilter(params.getLimitForQuery(), params.getOffset());

		StringBuilder sparqlQuery =
				new StringBuilder("SELECT DISTINCT ?rmapObjId ?type "
									+ "WHERE { "
									+ " GRAPH ?rmapObjId "
									+ "	  {"
									+ "		{" + sResourceIri + " <" + RDF.TYPE.toString() + "> ?type} ."
									+ "	  } . "
									+ " GRAPH ?eventId {"
									+ "   ?eventId <" + RDF.TYPE + "> <" + RMAP.EVENT + "> ."
									+ "   {?eventId <" + PROV.GENERATED + "> ?rmapObjId} UNION"
									+ "   {?eventId <" + RMAP.DERIVEDOBJECT + "> ?rmapObjId} .");

		if (dateFilterSparql.length()>0){
			sparqlQuery.append("   ?eventId <" + PROV.STARTEDATTIME + "> ?startDate .");
		}
		sparqlQuery.append(sysAgentSparql
							+ "  } "
							+ statusFilterSparql
							+ dateFilterSparql
							+ "} ");

		if (params.getOrderBy()==OrderBy.SELECT_ORDER){
			sparqlQuery.append("ORDER BY ?rmapObjId ?type");
		}
		sparqlQuery.append(limitOffsetFilterSparql);

		List<BindingSet> resultset = null;
		try {
			resultset = ts.getSPARQLQueryResults(sparqlQuery.toString());
		}
		catch (Exception e) {
			throw new RMapException("Could not retrieve SPARQL query results using " + sparqlQuery, e);
		}

		Map<IRI, Set<IRI>> map = null;

		try{
			if (resultset !=null && resultset.size()>0){
				map = new HashMap<IRI, Set<IRI>>();
				for (BindingSet bindingSet:resultset) {
					IRI rmapObjId = (IRI) bindingSet.getBinding("rmapObjId").getValue();
					Value typeVal = bindingSet.getBinding("type").getValue();
					if (!(typeVal instanceof IRI)){
						continue;
					}
					IRI type = (IRI)typeVal;
					if (map.containsKey(rmapObjId)){
						Set<IRI>types = map.get(rmapObjId);
						types.add(type);
					}
					else {
						Set<IRI>types = new HashSet<IRI>();
						types.add(type);
						map.put(rmapObjId, types);
					}
				}
				if (map.keySet().size()==0){
					map = null;
				}
			}
		}
		catch (RMapException r){throw r;}
		catch (Exception e){
			throw new RMapException("Could not process SPARQL results for resource's associated Objects", e);
		}

		return map;
	}

}


text:
import React from 'react';
import { ApiEndpoint, getRandomNumber } from '../CampusClient';

class UserProfileImage extends React.Component {
  state = {
    rnd: 0,
  };

  async componentDidMount() {
    this.setState({ rnd: getRandomNumber() });
  }

  render() {
    return (
      <img
        className="img-fluid"
        src={`${ApiEndpoint}Account/${this.props.user.id}/ProfileImage?tmp=${this.state.rnd}`}
        alt={this.props.user.fullName}
      />
    );
  }
}

export default UserProfileImage;

redpajama_set_name: RedPajamaGithub


text:
package ca.blarg.gdx.assets;

public class AssetLoadingException extends RuntimeException {
	String assetFilename;
	String detailMessage;

	public AssetLoadingException(String assetFilename, String message) {
		super(message);
		this.assetFilename = assetFilename;
		setDetailMessage(assetFilename, message);
	}

	public AssetLoadingException(String assetFilename, String message, Throwable cause) {
		super(message, cause);
		this.assetFilename = assetFilename;
		setDetailMessage(assetFilename, message);
	}

	private void setDetailMessage(String assetFilename, String message) {
		detailMessage = String.format("%s: %s", assetFilename, message);
	}

	@Override
	public String getMessage() {
		return detailMessage;
	}

	public String getAssetFilename() {
		return assetFilename;
	}
}


text:
module Octobat
  module Util
    def self.expand_nested_objects(h)
      case h
      when Hash
        res = {}
        h.each { |k, v| res[k] = expand_nested_objects(v) unless v.nil? }
        res
      when Array
        if !h.empty? && h.first.is_a?(Hash)
          res = {'' => h.map{|v| expand_nested_objects(v)}}
        else
          res = h
        end
        res
      else
        h
      end
    end

    def self.objects_to_ids(h)
      case h
      when APIResource
        h.id
      when Hash
        res = {}
        h.each { |k, v| res[k] = objects_to_ids(v) unless v.nil? }
        res
      when Array
        h.map { |v| objects_to_ids(v) }
      else
        h
      end
    end

    def self.object_classes
      @object_classes ||= {
        # data structures
        'list' => ListObject,

        # business objects
        'payment_mode' => PaymentMode,
        'payment' => Payment,
        'numbering_sequence' => NumberingSequence,
        'credit_note_numbering_sequence' => CreditNoteNumberingSequence,
        'invoice' => Invoice,
        'invoice_item' => InvoiceItem,
        'customer' => Customer
      }
    end

    def self.convert_to_octobat_object(resp, api_key)
      case resp
      when Array
        resp.map { |i| convert_to_octobat_object(i, api_key) }
      when Hash
        # Try converting to a known object class.  If none available, fall back to generic OctobatObject
        object_classes.fetch(resp[:object], OctobatObject).construct_from(resp, api_key)
      else
        resp
      end
    end

    def self.file_readable(file)
      # This is nominally equivalent to File.readable?, but that can
      # report incorrect results on some more oddball filesystems
      # (such as AFS)
      begin
        File.open(file) { |f| }
      rescue
        false
      else
        true
      end
    end

    def self.symbolize_names(object)
      case object
      when Hash
        new_hash = {}
        object.each do |key, value|
          key = (key.to_sym rescue key) || key
          new_hash[key] = symbolize_names(value)
        end
        new_hash
      when Array
        object.map { |value| symbolize_names(value) }
      else
        object
      end
    end

    def self.url_encode(key)
      URI.escape(key.to_s, Regexp.new("[^#{URI::PATTERN::UNRESERVED}]"))
    end

    def self.flatten_params(params, parent_key=nil)
      result = []
      params.each do |key, value|
        calculated_key = parent_key ? "#{parent_key}[#{url_encode(key)}]" : url_encode(key)
        if value.is_a?(Hash)
          result += flatten_params(value, calculated_key)
        elsif value.is_a?(Array)
          result += flatten_params_array(value, calculated_key)
        else
          result << [calculated_key, value]
        end
      end
      result
    end

    def self.flatten_params_array(value, calculated_key)
      result = []
      value.each do |elem|
        if elem.is_a?(Hash)
          result += flatten_params(elem, calculated_key)
        elsif elem.is_a?(Array)
          result += flatten_params_array(elem, calculated_key)
        else
          result << ["#{calculated_key}[]", elem]
        end
      end
      result
    end

    # The secondary opts argument can either be a string or hash
    # Turn this value into an api_key and a set of headers
    def self.parse_opts(opts)
      case opts
      when NilClass
        return nil, {}
      when String
        return opts, {}
      when Hash
        headers = {}
        if opts[:idempotency_key]
          headers[:idempotency_key] = opts[:idempotency_key]
        end
        if opts[:octobat_account]
          headers[:octobat_account] = opts[:octobat_account]
        end
        return opts[:api_key], headers
      else
        raise TypeError.new("parse_opts expects a string or a hash")
      end
    end
  end
end


text:
import { indexPhp } from '../support/constants';
import navigateTo from './navigateTo';

export default function formToUrl(e) {
  const form = new FormData(e.target);
  if (e.submitter?.name) {
    form.append(e.submitter.name, e.submitter.value);
  }
  navigateTo(`${indexPhp}?${(new URLSearchParams(form)).toString()}`);
}


text:
TEST(HandlerTest, testCompositeDecreeHandlerWithSingleHandler)
{
    bool was_called = false;
    paxos::CompositeHandler handler;
    handler.AddHandler([&was_called](std::string entry){ was_called=true; });

    ASSERT_FALSE(was_called);

    handler("some entry");

    ASSERT_TRUE(was_called);
}

TEST(HandlerTest, testCompositeDecreeHandlerWithMultipleHandlers)
{
    bool was_called = false;
    bool was_called_too = false;

    paxos::CompositeHandler handler;
    handler.AddHandler([&was_called](std::string entry){ was_called=true; });
    handler.AddHandler([&was_called_too](std::string entry){ was_called_too=true; });

    ASSERT_FALSE(was_called);
    ASSERT_FALSE(was_called_too);

    handler("some entry");

    ASSERT_TRUE(was_called);
    ASSERT_TRUE(was_called_too);
}

TEST(HandlerTest, testHandleAddReplicaSavesNewReplicaSet)
{
    auto replica = paxos::Replica("myhost", 8080);
    auto replicaset = std::make_shared<paxos::ReplicaSet>();
    replicaset->Add(replica);

    bool called_saved_replicaset = false;

    paxos::HandleAddReplica handler(
        "mylocation",
        replica,
        replicaset,
        std::make_shared<paxos::Signal>(),
        [&called_saved_replicaset](
            std::shared_ptr<paxos::ReplicaSet>,
            std::ostream&)
        {
            called_saved_replicaset = true;
        }
    );
    handler(
        paxos::Serialize<paxos::UpdateReplicaSetDecree>(
            {
                paxos::Replica("author", 1010), // author
                paxos::Replica("yourhost", 8080),    // replica
                "remote_directory"
            }
        )
    );

    ASSERT_TRUE(called_saved_replicaset);
    ASSERT_TRUE(replicaset->Contains(paxos::Replica("yourhost", 8080)));
}

TEST(HandlerTest, testHandleRemoveReplicaSavesNewReplicaSet)
{
    auto replica = paxos::Replica("myhost", 8080);
    auto replicaset = std::make_shared<paxos::ReplicaSet>();
    replicaset->Add(replica);

    bool called_saved_replicaset = false;

    paxos::HandleRemoveReplica handler(
        "mylocation",
        replica,
        replicaset,
        std::make_shared<paxos::Signal>(),
        [&called_saved_replicaset](
            std::shared_ptr<paxos::ReplicaSet>,
            std::ostream&)
        {
            called_saved_replicaset = true;
        }
    );
    handler(
        paxos::Serialize<paxos::UpdateReplicaSetDecree>(
            {
                paxos::Replica("author", 1010), // author
                paxos::Replica("myhost", 8080),    // replica
                "remote_directory"
            }
        )
    );

    ASSERT_TRUE(called_saved_replicaset);
    ASSERT_FALSE(replicaset->Contains(replica));
}


text:
package annis.gui;

import annis.libgui.Helper;
import com.vaadin.ui.Button.ClickEvent;
import com.vaadin.ui.*;
import com.vaadin.ui.themes.ChameleonTheme;
import java.util.Set;

/**
 *
 * @author thomas
 */
public class CitationWindow extends Window
  implements Button.ClickListener
{

  public CitationWindow( String query, Set<String> corpora,
    int contextLeft, int contextRight)
  {
    super("Citation");

    VerticalLayout wLayout = new VerticalLayout();
    setContent(wLayout);
    wLayout.setSizeFull();

    String url = Helper.generateCitation(query, corpora, contextLeft,
      contextRight, null, 0, 10);

    TextArea txtCitation = new TextArea();

    txtCitation.setWidth("100%");
    txtCitation.setHeight("100%");
    txtCitation.addStyleName(ChameleonTheme.TEXTFIELD_BIG);
    txtCitation.addStyleName("citation");
    txtCitation.setValue(url);
    txtCitation.setWordwrap(true);
    txtCitation.setReadOnly(true);

    wLayout.addComponent(txtCitation);

    Button btOk = new Button("OK");
    btOk.addListener((Button.ClickListener) this);
    btOk.setSizeUndefined();

    wLayout.addComponent(btOk);

    wLayout.setExpandRatio(txtCitation, 1.0f);
    wLayout.setComponentAlignment(btOk, Alignment.BOTTOM_CENTER);

    setWidth("400px");
    setHeight("200px");

  }

  @Override
  public void buttonClick(ClickEvent event)
  {
    UI.getCurrent().removeWindow(this);
  }
}


text:
    var camera, scene, renderer;
    var effect, controls;
    var element, container;

    var clock = new THREE.Clock();

    var cardboard = THREE.Cardboard;

    init();
    animate();

    function init() {
      renderer = new THREE.WebGLRenderer();
      element = renderer.domElement;
      container = document.getElementById('scene');
      container.appendChild(element);

      scene = new THREE.Scene();

      camera = new THREE.PerspectiveCamera(90, 1, 0.001, 700);
      camera.position.set(0, 10, 0);
      scene.add(camera);

      cardboard.init({
        renderer: renderer,
        camera: camera,
        scene: scene
      });

      controls = new THREE.OrbitControls(camera, element);
      controls.rotateUp(Math.PI / 4);
      controls.target.set(
        camera.position.x + 0.1,
        camera.position.y,
        camera.position.z
      );
      controls.noZoom = true;
      controls.noPan = true;

      var light = new THREE.HemisphereLight(0x777777, 0x000000, 0.6);
      scene.add(light);

      var texture = THREE.ImageUtils.loadTexture(
        'textures/patterns/checker.png'
      );
      texture.wrapS = THREE.RepeatWrapping;
      texture.wrapT = THREE.RepeatWrapping;
      texture.repeat = new THREE.Vector2(50, 50);
      texture.anisotropy = renderer.getMaxAnisotropy();

      var material = new THREE.MeshPhongMaterial({
        color: 0xffffff,
        specular: 0xffffff,
        shininess: 20,
        shading: THREE.FlatShading,
        map: texture
      });

      var geometry = new THREE.PlaneGeometry(1000, 1000);

      var mesh = new THREE.Mesh(geometry, material);
      mesh.rotation.x = -Math.PI / 2;
      scene.add(mesh);

      window.addEventListener('resize', resize, false);
      setTimeout(resize, 1);
    }

    function update(deltaTime) {
      resize();
      controls.update(deltaTime);
      cardboard.update();
    }

    function render(deltaTime) {
      cardboard.render();
    }

    function resize() {
      var width = container.offsetWidth;
      var height = container.offsetHeight;
      cardboard.resize(width, height)
    }

    function animate(time) {
      requestAnimationFrame(animate);
      update(clock.getDelta());
      render(clock.getDelta());
    }


text:
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;

namespace Foundatio.Repositories.Migrations {
    public abstract class MigrationBase : IMigration {
        protected ILogger _logger;

        public MigrationBase(ILoggerFactory loggerFactory = null) {
            _logger = loggerFactory?.CreateLogger(GetType()) ?? NullLogger.Instance;
        }

        public virtual int? Version { get; } = null;
        public abstract Task RunAsync();

        protected int CalculateProgress(long total, long completed, int startProgress = 0, int endProgress = 100) {
            return startProgress + (int)((100 * (double)completed / total) * (((double)endProgress - startProgress) / 100));
        }
    }
}


text:
<?php

namespace Magento\Setup\Test\Unit\Module\Dependency\Report\Dependency\Data;

use Magento\Framework\TestFramework\Unit\Helper\ObjectManager;

class ModuleTest extends \PHPUnit_Framework_TestCase
{
    /**
     * @var \Magento\Setup\Module\Dependency\Report\Dependency\Data\Dependency|\PHPUnit_Framework_MockObject_MockObject
     */
    protected $dependencyFirst;

    /**
     * @var \Magento\Setup\Module\Dependency\Report\Dependency\Data\Dependency|\PHPUnit_Framework_MockObject_MockObject
     */
    protected $dependencySecond;

    /**
     * @var \Magento\Setup\Module\Dependency\Report\Dependency\Data\Module
     */
    protected $module;

    public function setUp()
    {
        $this->dependencyFirst = $this->getMock(
            'Magento\Setup\Module\Dependency\Report\Dependency\Data\Dependency',
            [],
            [],
            '',
            false
        );
        $this->dependencySecond = $this->getMock(
            'Magento\Setup\Module\Dependency\Report\Dependency\Data\Dependency',
            [],
            [],
            '',
            false
        );

        $objectManagerHelper = new ObjectManager($this);
        $this->module = $objectManagerHelper->getObject(
            'Magento\Setup\Module\Dependency\Report\Dependency\Data\Module',
            ['name' => 'name', 'dependencies' => [$this->dependencyFirst, $this->dependencySecond]]
        );
    }

    public function testGetName()
    {
        $this->assertEquals('name', $this->module->getName());
    }

    public function testGetDependencies()
    {
        $this->assertEquals([$this->dependencyFirst, $this->dependencySecond], $this->module->getDependencies());
    }

    public function testGetDependenciesCount()
    {
        $this->assertEquals(2, $this->module->getDependenciesCount());
    }

    public function testGetHardDependenciesCount()
    {
        $this->dependencyFirst->expects($this->once())->method('isHard')->will($this->returnValue(true));
        $this->dependencyFirst->expects($this->never())->method('isSoft');

        $this->dependencySecond->expects($this->once())->method('isHard')->will($this->returnValue(false));
        $this->dependencySecond->expects($this->never())->method('isSoft');

        $this->assertEquals(1, $this->module->getHardDependenciesCount());
    }

    public function testGetSoftDependenciesCount()
    {
        $this->dependencyFirst->expects($this->never())->method('isHard');
        $this->dependencyFirst->expects($this->once())->method('isSoft')->will($this->returnValue(true));

        $this->dependencySecond->expects($this->never())->method('isHard');
        $this->dependencySecond->expects($this->once())->method('isSoft')->will($this->returnValue(false));

        $this->assertEquals(1, $this->module->getSoftDependenciesCount());
    }
}


text:
"""Code for communicating with the Workers."""

from __future__ import absolute_import

import collections
import contextlib
import copy
import logging
import queue
import subprocess
import sys
import threading
import time
from builtins import object
from typing import Callable
from typing import DefaultDict
from typing import Dict
from typing import Iterable
from typing import Iterator
from typing import List
from typing import Mapping
from typing import Optional
from typing import Tuple
from typing import Union
from typing import cast
from typing import overload

import grpc

from apache_beam.io import filesystems
from apache_beam.portability import common_urns
from apache_beam.portability import python_urns
from apache_beam.portability.api import beam_artifact_api_pb2
from apache_beam.portability.api import beam_artifact_api_pb2_grpc
from apache_beam.portability.api import beam_fn_api_pb2
from apache_beam.portability.api import beam_fn_api_pb2_grpc
from apache_beam.portability.api import beam_provision_api_pb2
from apache_beam.portability.api import beam_provision_api_pb2_grpc
from apache_beam.portability.api import beam_runner_api_pb2
from apache_beam.portability.api import endpoints_pb2
from apache_beam.runners.portability import artifact_service
from apache_beam.runners.portability.fn_api_runner.execution import Buffer
from apache_beam.runners.worker import data_plane
from apache_beam.runners.worker import sdk_worker
from apache_beam.runners.worker.channel_factory import GRPCChannelFactory
from apache_beam.runners.worker.sdk_worker import _Future
from apache_beam.runners.worker.statecache import StateCache
from apache_beam.utils import proto_utils
from apache_beam.utils import thread_pool_executor

# State caching is enabled in the fn_api_runner for testing, except for one
# test which runs without state caching (FnApiRunnerTestWithDisabledCaching).
# The cache is disabled in production for other runners.
STATE_CACHE_SIZE = 100

# Time-based flush is enabled in the fn_api_runner by default.
DATA_BUFFER_TIME_LIMIT_MS = 1000

_LOGGER = logging.getLogger(__name__)

ConstructorFn = Callable[[
    Union['message.Message', bytes],
    'StateServicer',
    Optional['ExtendedProvisionInfo'],
    'GrpcServer'
],
                         'WorkerHandler']

class ControlConnection(object):

  _uid_counter = 0
  _lock = threading.Lock()

  def __init__(self):
    self._push_queue = queue.Queue(
    )  # type: queue.Queue[beam_fn_api_pb2.InstructionRequest]
    self._input = None  # type: Optional[Iterable[beam_fn_api_pb2.InstructionResponse]]
    self._futures_by_id = dict()  # type: Dict[str, ControlFuture]
    self._read_thread = threading.Thread(
        name='beam_control_read', target=self._read)
    self._state = BeamFnControlServicer.UNSTARTED_STATE

  def _read(self):
    for data in self._input:
      self._futures_by_id.pop(data.instruction_id).set(data)

  @overload
  def push(self, req):
    # type: (BeamFnControlServicer.DoneMarker) -> None
    pass

  @overload
  def push(self, req):
    # type: (beam_fn_api_pb2.InstructionRequest) -> ControlFuture
    pass

  def push(self, req):
    if req == BeamFnControlServicer._DONE_MARKER:
      self._push_queue.put(req)
      return None
    if not req.instruction_id:
      with ControlConnection._lock:
        ControlConnection._uid_counter += 1
        req.instruction_id = 'control_%s' % ControlConnection._uid_counter
    future = ControlFuture(req.instruction_id)
    self._futures_by_id[req.instruction_id] = future
    self._push_queue.put(req)
    return future

  def get_req(self):
    # type: () -> beam_fn_api_pb2.InstructionRequest
    return self._push_queue.get()

  def set_input(self, input):
    # type: (Iterable[beam_fn_api_pb2.InstructionResponse]) -> None
    with ControlConnection._lock:
      if self._input:
        raise RuntimeError('input is already set.')
      self._input = input
      self._read_thread.start()
      self._state = BeamFnControlServicer.STARTED_STATE

  def close(self):
    # type: () -> None
    with ControlConnection._lock:
      if self._state == BeamFnControlServicer.STARTED_STATE:
        self.push(BeamFnControlServicer._DONE_MARKER)
        self._read_thread.join()
      self._state = BeamFnControlServicer.DONE_STATE

  def abort(self, exn):
    for future in self._futures_by_id.values():
      future.abort(exn)

class BeamFnControlServicer(beam_fn_api_pb2_grpc.BeamFnControlServicer):
  """Implementation of BeamFnControlServicer for clients."""

  UNSTARTED_STATE = 'unstarted'
  STARTED_STATE = 'started'
  DONE_STATE = 'done'

  class DoneMarker(object):
    pass

  _DONE_MARKER = DoneMarker()

  def __init__(
      self,
      worker_manager,  # type: WorkerHandlerManager
  ):
    self._worker_manager = worker_manager
    self._lock = threading.Lock()
    self._uid_counter = 0
    self._state = self.UNSTARTED_STATE
    # following self._req_* variables are used for debugging purpose, data is
    # added only when self._log_req is True.
    self._req_sent = collections.defaultdict(int)
    self._req_worker_mapping = {}
    self._log_req = logging.getLogger().getEffectiveLevel() <= logging.DEBUG
    self._connections_by_worker_id = collections.defaultdict(
        ControlConnection)  # type: DefaultDict[str, ControlConnection]

  def get_conn_by_worker_id(self, worker_id):
    # type: (str) -> ControlConnection
    with self._lock:
      return self._connections_by_worker_id[worker_id]

  def Control(self,
              iterator,  # type: Iterable[beam_fn_api_pb2.InstructionResponse]
              context
             ):
    # type: (...) -> Iterator[beam_fn_api_pb2.InstructionRequest]
    with self._lock:
      if self._state == self.DONE_STATE:
        return
      else:
        self._state = self.STARTED_STATE

    worker_id = dict(context.invocation_metadata()).get('worker_id')
    if not worker_id:
      raise RuntimeError(
          'All workers communicate through gRPC should have '
          'worker_id. Received None.')

    control_conn = self.get_conn_by_worker_id(worker_id)
    control_conn.set_input(iterator)

    while True:
      to_push = control_conn.get_req()
      if to_push is self._DONE_MARKER:
        return
      yield to_push
      if self._log_req:
        self._req_sent[to_push.instruction_id] += 1

  def done(self):
    self._state = self.DONE_STATE
    _LOGGER.debug(
        'Runner: Requests sent by runner: %s',
        [(str(req), cnt) for req, cnt in self._req_sent.items()])
    _LOGGER.debug(
        'Runner: Requests multiplexing info: %s',
        [(str(req), worker)
         for req, worker in self._req_worker_mapping.items()])

  def GetProcessBundleDescriptor(self, id, context=None):
    return self._worker_manager.get_process_bundle_descriptor(id)

class WorkerHandler(object):
  """worker_handler for a worker.

  It provides utilities to start / stop the worker, provision any resources for
  it, as well as provide descriptors for the data, state and logging APIs for
  it.
  """

  _registered_environments = {}  # type: Dict[str, Tuple[ConstructorFn, type]]
  _worker_id_counter = -1
  _lock = threading.Lock()

  control_conn = None  # type: ControlConnection
  data_conn = None  # type: data_plane._GrpcDataChannel

  def __init__(self,
               control_handler,
               data_plane_handler,
               state,  # type: StateServicer
               provision_info  # type: Optional[ExtendedProvisionInfo]
              ):
    # type: (...) -> None

    """Initialize a WorkerHandler.

    Args:
      control_handler:
      data_plane_handler (data_plane.DataChannel):
      state:
      provision_info:
    """
    self.control_handler = control_handler
    self.data_plane_handler = data_plane_handler
    self.state = state
    self.provision_info = provision_info

    with WorkerHandler._lock:
      WorkerHandler._worker_id_counter += 1
      self.worker_id = 'worker_%s' % WorkerHandler._worker_id_counter

  def close(self):
    # type: () -> None
    self.stop_worker()

  def start_worker(self):
    # type: () -> None
    raise NotImplementedError

  def stop_worker(self):
    # type: () -> None
    raise NotImplementedError

  def data_api_service_descriptor(self):
    # type: () -> Optional[endpoints_pb2.ApiServiceDescriptor]
    raise NotImplementedError

  def state_api_service_descriptor(self):
    # type: () -> Optional[endpoints_pb2.ApiServiceDescriptor]
    raise NotImplementedError

  def logging_api_service_descriptor(self):
    # type: () -> Optional[endpoints_pb2.ApiServiceDescriptor]
    raise NotImplementedError

  @classmethod
  def register_environment(
      cls,
      urn,  # type: str
      payload_type  # type: Optional[Type[T]]
  ):
    # type: (...) -> Callable[[Callable[[T, StateServicer, Optional[ExtendedProvisionInfo], GrpcServer], WorkerHandler]], Callable[[T, StateServicer, Optional[ExtendedProvisionInfo], GrpcServer], WorkerHandler]]
    def wrapper(constructor):
      cls._registered_environments[urn] = constructor, payload_type
      return constructor

    return wrapper

  @classmethod
  def create(cls,
             environment,  # type: beam_runner_api_pb2.Environment
             state,  # type: StateServicer
             provision_info,  # type: Optional[ExtendedProvisionInfo]
             grpc_server  # type: GrpcServer
            ):
    # type: (...) -> WorkerHandler
    constructor, payload_type = cls._registered_environments[environment.urn]
    return constructor(
        proto_utils.parse_Bytes(environment.payload, payload_type),
        state,
        provision_info,
        grpc_server)

@WorkerHandler.register_environment(python_urns.EMBEDDED_PYTHON, None)
class EmbeddedWorkerHandler(WorkerHandler):
  """An in-memory worker_handler for fn API control, state and data planes."""

  def __init__(self,
               unused_payload,  # type: None
               state,  # type: sdk_worker.StateHandler
               provision_info,  # type: Optional[ExtendedProvisionInfo]
               worker_manager,  # type: WorkerHandlerManager
              ):
    # type: (...) -> None
    super(EmbeddedWorkerHandler, self).__init__(
        self, data_plane.InMemoryDataChannel(), state, provision_info)
    self.control_conn = self  # type: ignore  # need Protocol to describe this
    self.data_conn = self.data_plane_handler
    state_cache = StateCache(STATE_CACHE_SIZE)
    self.bundle_processor_cache = sdk_worker.BundleProcessorCache(
        SingletonStateHandlerFactory(
            sdk_worker.CachingStateHandler(state_cache, state)),
        data_plane.InMemoryDataChannelFactory(
            self.data_plane_handler.inverse()),
        worker_manager._process_bundle_descriptors)
    self.worker = sdk_worker.SdkWorker(
        self.bundle_processor_cache,
        state_cache_metrics_fn=state_cache.get_monitoring_infos)
    self._uid_counter = 0

  def push(self, request):
    if not request.instruction_id:
      self._uid_counter += 1
      request.instruction_id = 'control_%s' % self._uid_counter
    response = self.worker.do_instruction(request)
    return ControlFuture(request.instruction_id, response)

  def start_worker(self):
    # type: () -> None
    pass

  def stop_worker(self):
    # type: () -> None
    self.bundle_processor_cache.shutdown()

  def done(self):
    # type: () -> None
    pass

  def data_api_service_descriptor(self):
    # type: () -> None
    return None

  def state_api_service_descriptor(self):
    # type: () -> None
    return None

  def logging_api_service_descriptor(self):
    # type: () -> None
    return None

class BasicLoggingService(beam_fn_api_pb2_grpc.BeamFnLoggingServicer):

  LOG_LEVEL_MAP = {
      beam_fn_api_pb2.LogEntry.Severity.CRITICAL: logging.CRITICAL,
      beam_fn_api_pb2.LogEntry.Severity.ERROR: logging.ERROR,
      beam_fn_api_pb2.LogEntry.Severity.WARN: logging.WARNING,
      beam_fn_api_pb2.LogEntry.Severity.NOTICE: logging.INFO + 1,
      beam_fn_api_pb2.LogEntry.Severity.INFO: logging.INFO,
      beam_fn_api_pb2.LogEntry.Severity.DEBUG: logging.DEBUG,
      beam_fn_api_pb2.LogEntry.Severity.TRACE: logging.DEBUG - 1,
      beam_fn_api_pb2.LogEntry.Severity.UNSPECIFIED: logging.NOTSET,
  }

  def Logging(self, log_messages, context=None):
    yield beam_fn_api_pb2.LogControl()
    for log_message in log_messages:
      for log in log_message.log_entries:
        logging.log(self.LOG_LEVEL_MAP[log.severity], str(log))

class BasicProvisionService(beam_provision_api_pb2_grpc.ProvisionServiceServicer
                            ):
  def __init__(self, base_info, worker_manager):
    # type: (Optional[beam_provision_api_pb2.ProvisionInfo], WorkerHandlerManager) -> None
    self._base_info = base_info
    self._worker_manager = worker_manager

  def GetProvisionInfo(self, request, context=None):
    # type: (...) -> beam_provision_api_pb2.GetProvisionInfoResponse
    if context:
      worker_id = dict(context.invocation_metadata())['worker_id']
      worker = self._worker_manager.get_worker(worker_id)
      info = copy.copy(worker.provision_info.provision_info)
      info.logging_endpoint.CopyFrom(worker.logging_api_service_descriptor())
      info.artifact_endpoint.CopyFrom(worker.artifact_api_service_descriptor())
      info.control_endpoint.CopyFrom(worker.control_api_service_descriptor())
    else:
      info = self._base_info
    return beam_provision_api_pb2.GetProvisionInfoResponse(info=info)

class EmptyArtifactRetrievalService(
    beam_artifact_api_pb2_grpc.LegacyArtifactRetrievalServiceServicer):
  def GetManifest(self, request, context=None):
    return beam_artifact_api_pb2.GetManifestResponse(
        manifest=beam_artifact_api_pb2.Manifest())

  def GetArtifact(self, request, context=None):
    raise ValueError('No artifacts staged.')

class GrpcServer(object):

  _DEFAULT_SHUTDOWN_TIMEOUT_SECS = 5

  def __init__(self,
               state,  # type: StateServicer
               provision_info,  # type: Optional[ExtendedProvisionInfo]
               worker_manager,  # type: WorkerHandlerManager
              ):
    # type: (...) -> None
    self.state = state
    self.provision_info = provision_info
    self.control_server = grpc.server(
        thread_pool_executor.shared_unbounded_instance())
    self.control_port = self.control_server.add_insecure_port('[::]:0')
    self.control_address = 'localhost:%s' % self.control_port

    # Options to have no limits (-1) on the size of the messages
    # received or sent over the data plane. The actual buffer size
    # is controlled in a layer above.
    no_max_message_sizes = [("grpc.max_receive_message_length", -1),
                            ("grpc.max_send_message_length", -1)]
    self.data_server = grpc.server(
        thread_pool_executor.shared_unbounded_instance(),
        options=no_max_message_sizes)
    self.data_port = self.data_server.add_insecure_port('[::]:0')

    self.state_server = grpc.server(
        thread_pool_executor.shared_unbounded_instance(),
        options=no_max_message_sizes)
    self.state_port = self.state_server.add_insecure_port('[::]:0')

    self.control_handler = BeamFnControlServicer(worker_manager)
    beam_fn_api_pb2_grpc.add_BeamFnControlServicer_to_server(
        self.control_handler, self.control_server)

    # If we have provision info, serve these off the control port as well.
    if self.provision_info:
      if self.provision_info.provision_info:
        beam_provision_api_pb2_grpc.add_ProvisionServiceServicer_to_server(
            BasicProvisionService(
                self.provision_info.provision_info, worker_manager),
            self.control_server)

      if self.provision_info.artifact_staging_dir:
        service = artifact_service.BeamFilesystemArtifactService(
            self.provision_info.artifact_staging_dir
        )  # type: beam_artifact_api_pb2_grpc.LegacyArtifactRetrievalServiceServicer
      else:
        service = EmptyArtifactRetrievalService()
      beam_artifact_api_pb2_grpc.add_LegacyArtifactRetrievalServiceServicer_to_server(
          service, self.control_server)

      beam_artifact_api_pb2_grpc.add_ArtifactRetrievalServiceServicer_to_server(
          artifact_service.ArtifactRetrievalService(
              file_reader=filesystems.FileSystems.open),
          self.control_server)

    self.data_plane_handler = data_plane.BeamFnDataServicer(
        DATA_BUFFER_TIME_LIMIT_MS)
    beam_fn_api_pb2_grpc.add_BeamFnDataServicer_to_server(
        self.data_plane_handler, self.data_server)

    beam_fn_api_pb2_grpc.add_BeamFnStateServicer_to_server(
        GrpcStateServicer(state), self.state_server)

    self.logging_server = grpc.server(
        thread_pool_executor.shared_unbounded_instance(),
        options=no_max_message_sizes)
    self.logging_port = self.logging_server.add_insecure_port('[::]:0')
    beam_fn_api_pb2_grpc.add_BeamFnLoggingServicer_to_server(
        BasicLoggingService(), self.logging_server)

    _LOGGER.info('starting control server on port %s', self.control_port)
    _LOGGER.info('starting data server on port %s', self.data_port)
    _LOGGER.info('starting state server on port %s', self.state_port)
    _LOGGER.info('starting logging server on port %s', self.logging_port)
    self.logging_server.start()
    self.state_server.start()
    self.data_server.start()
    self.control_server.start()

  def close(self):
    self.control_handler.done()
    to_wait = [
        self.control_server.stop(self._DEFAULT_SHUTDOWN_TIMEOUT_SECS),
        self.data_server.stop(self._DEFAULT_SHUTDOWN_TIMEOUT_SECS),
        self.state_server.stop(self._DEFAULT_SHUTDOWN_TIMEOUT_SECS),
        self.logging_server.stop(self._DEFAULT_SHUTDOWN_TIMEOUT_SECS)
    ]
    for w in to_wait:
      w.wait()

class GrpcWorkerHandler(WorkerHandler):
  """An grpc based worker_handler for fn API control, state and data planes."""

  def __init__(self,
               state,  # type: StateServicer
               provision_info,  # type: Optional[ExtendedProvisionInfo]
               grpc_server  # type: GrpcServer
              ):
    # type: (...) -> None
    self._grpc_server = grpc_server
    super(GrpcWorkerHandler, self).__init__(
        self._grpc_server.control_handler,
        self._grpc_server.data_plane_handler,
        state,
        provision_info)
    self.state = state

    self.control_address = self.port_from_worker(self._grpc_server.control_port)
    self.control_conn = self._grpc_server.control_handler.get_conn_by_worker_id(
        self.worker_id)

    self.data_conn = self._grpc_server.data_plane_handler.get_conn_by_worker_id(
        self.worker_id)

  def control_api_service_descriptor(self):
    # type: () -> endpoints_pb2.ApiServiceDescriptor
    return endpoints_pb2.ApiServiceDescriptor(
        url=self.port_from_worker(self._grpc_server.control_port))

  def artifact_api_service_descriptor(self):
    # type: () -> endpoints_pb2.ApiServiceDescriptor
    return endpoints_pb2.ApiServiceDescriptor(
        url=self.port_from_worker(self._grpc_server.control_port))

  def data_api_service_descriptor(self):
    # type: () -> endpoints_pb2.ApiServiceDescriptor
    return endpoints_pb2.ApiServiceDescriptor(
        url=self.port_from_worker(self._grpc_server.data_port))

  def state_api_service_descriptor(self):
    # type: () -> endpoints_pb2.ApiServiceDescriptor
    return endpoints_pb2.ApiServiceDescriptor(
        url=self.port_from_worker(self._grpc_server.state_port))

  def logging_api_service_descriptor(self):
    # type: () -> endpoints_pb2.ApiServiceDescriptor
    return endpoints_pb2.ApiServiceDescriptor(
        url=self.port_from_worker(self._grpc_server.logging_port))

  def close(self):
    # type: () -> None
    self.control_conn.close()
    self.data_conn.close()
    super(GrpcWorkerHandler, self).close()

  def port_from_worker(self, port):
    return '%s:%s' % (self.host_from_worker(), port)

  def host_from_worker(self):
    return 'localhost'

@WorkerHandler.register_environment(
    common_urns.environments.EXTERNAL.urn, beam_runner_api_pb2.ExternalPayload)
class ExternalWorkerHandler(GrpcWorkerHandler):
  def __init__(self,
               external_payload,  # type: beam_runner_api_pb2.ExternalPayload
               state,  # type: StateServicer
               provision_info,  # type: Optional[ExtendedProvisionInfo]
               grpc_server  # type: GrpcServer
              ):
    # type: (...) -> None
    super(ExternalWorkerHandler,
          self).__init__(state, provision_info, grpc_server)
    self._external_payload = external_payload

  def start_worker(self):
    # type: () -> None
    _LOGGER.info("Requesting worker at %s", self._external_payload.endpoint.url)
    stub = beam_fn_api_pb2_grpc.BeamFnExternalWorkerPoolStub(
        GRPCChannelFactory.insecure_channel(
            self._external_payload.endpoint.url))
    control_descriptor = endpoints_pb2.ApiServiceDescriptor(
        url=self.control_address)
    response = stub.StartWorker(
        beam_fn_api_pb2.StartWorkerRequest(
            worker_id=self.worker_id,
            control_endpoint=control_descriptor,
            artifact_endpoint=control_descriptor,
            provision_endpoint=control_descriptor,
            logging_endpoint=self.logging_api_service_descriptor(),
            params=self._external_payload.params))
    if response.error:
      raise RuntimeError("Error starting worker: %s" % response.error)

  def stop_worker(self):
    # type: () -> None
    pass

  def host_from_worker(self):
    # TODO(BEAM-8646): Reconcile across platforms.
    if sys.platform in ['win32', 'darwin']:
      return 'localhost'
    import socket
    return socket.getfqdn()

@WorkerHandler.register_environment(python_urns.EMBEDDED_PYTHON_GRPC, bytes)
class EmbeddedGrpcWorkerHandler(GrpcWorkerHandler):
  def __init__(self,
               payload,  # type: bytes
               state,  # type: StateServicer
               provision_info,  # type: Optional[ExtendedProvisionInfo]
               grpc_server  # type: GrpcServer
              ):
    # type: (...) -> None
    super(EmbeddedGrpcWorkerHandler,
          self).__init__(state, provision_info, grpc_server)

    from apache_beam.transforms.environments import EmbeddedPythonGrpcEnvironment
    config = EmbeddedPythonGrpcEnvironment.parse_config(payload.decode('utf-8'))
    self._state_cache_size = config.get('state_cache_size') or STATE_CACHE_SIZE
    self._data_buffer_time_limit_ms = \
        config.get('data_buffer_time_limit_ms') or DATA_BUFFER_TIME_LIMIT_MS

  def start_worker(self):
    # type: () -> None
    self.worker = sdk_worker.SdkHarness(
        self.control_address,
        state_cache_size=self._state_cache_size,
        data_buffer_time_limit_ms=self._data_buffer_time_limit_ms,
        worker_id=self.worker_id)
    self.worker_thread = threading.Thread(
        name='run_worker', target=self.worker.run)
    self.worker_thread.daemon = True
    self.worker_thread.start()

  def stop_worker(self):
    # type: () -> None
    self.worker_thread.join()

# The subprocesses module is not threadsafe on Python 2.7. Use this lock to
# prevent concurrent calls to POpen().
SUBPROCESS_LOCK = threading.Lock()

@WorkerHandler.register_environment(python_urns.SUBPROCESS_SDK, bytes)
class SubprocessSdkWorkerHandler(GrpcWorkerHandler):
  def __init__(self,
               worker_command_line,  # type: bytes
               state,  # type: StateServicer
               provision_info,  # type: Optional[ExtendedProvisionInfo]
               grpc_server  # type: GrpcServer
              ):
    # type: (...) -> None
    super(SubprocessSdkWorkerHandler,
          self).__init__(state, provision_info, grpc_server)
    self._worker_command_line = worker_command_line

  def start_worker(self):
    # type: () -> None
    from apache_beam.runners.portability import local_job_service
    self.worker = local_job_service.SubprocessSdkWorker(
        self._worker_command_line, self.control_address, self.worker_id)
    self.worker_thread = threading.Thread(
        name='run_worker', target=self.worker.run)
    self.worker_thread.start()

  def stop_worker(self):
    # type: () -> None
    self.worker_thread.join()

@WorkerHandler.register_environment(
    common_urns.environments.DOCKER.urn, beam_runner_api_pb2.DockerPayload)
class DockerSdkWorkerHandler(GrpcWorkerHandler):
  def __init__(self,
               payload,  # type: beam_runner_api_pb2.DockerPayload
               state,  # type: StateServicer
               provision_info,  # type: Optional[ExtendedProvisionInfo]
               grpc_server  # type: GrpcServer
              ):
    # type: (...) -> None
    super(DockerSdkWorkerHandler,
          self).__init__(state, provision_info, grpc_server)
    self._container_image = payload.container_image
    self._container_id = None  # type: Optional[bytes]

  def host_from_worker(self):
    if sys.platform == "darwin":
      # See https://docs.docker.com/docker-for-mac/networking/
      return 'host.docker.internal'
    else:
      return super(DockerSdkWorkerHandler, self).host_from_worker()

  def start_worker(self):
    # type: () -> None
    with SUBPROCESS_LOCK:
      try:
        subprocess.check_call(['docker', 'pull', self._container_image])
      except Exception:
        _LOGGER.info('Unable to pull image %s' % self._container_image)
      self._container_id = subprocess.check_output([
          'docker',
          'run',
          '-d',
          # TODO:  credentials
          '--network=host',
          self._container_image,
          '--id=%s' % self.worker_id,
          '--logging_endpoint=%s' % self.logging_api_service_descriptor().url,
          '--control_endpoint=%s' % self.control_address,
          '--artifact_endpoint=%s' % self.control_address,
          '--provision_endpoint=%s' % self.control_address,
      ]).strip()
      assert self._container_id is not None
      while True:
        status = subprocess.check_output([
            'docker', 'inspect', '-f', '{{.State.Status}}', self._container_id
        ]).strip()
        _LOGGER.info(
            'Waiting for docker to start up.Current status is %s' %
            status.decode('utf-8'))
        if status == b'running':
          _LOGGER.info(
              'Docker container is running. container_id = %s, '
              'worker_id = %s',
              self._container_id,
              self.worker_id)
          break
        elif status in (b'dead', b'exited'):
          subprocess.call(['docker', 'container', 'logs', self._container_id])
          raise RuntimeError(
              'SDK failed to start. Final status is %s' %
              status.decode('utf-8'))
      time.sleep(1)
    self._done = False
    t = threading.Thread(target=self.watch_container)
    t.daemon = True
    t.start()

  def watch_container(self):
    while not self._done:
      status = subprocess.check_output(
          ['docker', 'inspect', '-f', '{{.State.Status}}',
           self._container_id]).strip()
      if status != b'running':
        if not self._done:
          logs = subprocess.check_output([
              'docker', 'container', 'logs', '--tail', '10', self._container_id
          ],
                                         stderr=subprocess.STDOUT)
          _LOGGER.info(logs)
          self.control_conn.abort(
              RuntimeError(
                  'SDK exited unexpectedly. '
                  'Final status is %s. Final log line is %s' % (
                      status.decode('utf-8'),
                      logs.decode('utf-8').strip().split('\n')[-1])))
      time.sleep(5)

  def stop_worker(self):
    # type: () -> None
    self._done = True
    if self._container_id:
      with SUBPROCESS_LOCK:
        subprocess.call(['docker', 'kill', self._container_id])

class WorkerHandlerManager(object):
  """
  Manages creation of ``WorkerHandler``s.

  Caches ``WorkerHandler``s based on environment id.
  """
  def __init__(self,
               environments,  # type: Mapping[str, beam_runner_api_pb2.Environment]
               job_provision_info  # type: Optional[ExtendedProvisionInfo]
              ):
    # type: (...) -> None
    self._environments = environments
    self._job_provision_info = job_provision_info
    self._cached_handlers = collections.defaultdict(
        list)  # type: DefaultDict[str, List[WorkerHandler]]
    self._workers_by_id = {}  # type: Dict[str, WorkerHandler]
    self.state_servicer = StateServicer()
    self._grpc_server = None  # type: Optional[GrpcServer]
    self._process_bundle_descriptors = {}

  def register_process_bundle_descriptor(self, process_bundle_descriptor):
    self._process_bundle_descriptors[
        process_bundle_descriptor.id] = process_bundle_descriptor

  def get_process_bundle_descriptor(self, request):
    return self._process_bundle_descriptors[
        request.process_bundle_descriptor_id]

  def get_worker_handlers(
      self,
      environment_id,  # type: Optional[str]
      num_workers  # type: int
  ):
    # type: (...) -> List[WorkerHandler]
    if environment_id is None:
      # Any environment will do, pick one arbitrarily.
      environment_id = next(iter(self._environments.keys()))
    environment = self._environments[environment_id]

    # assume all environments except EMBEDDED_PYTHON use gRPC.
    if environment.urn == python_urns.EMBEDDED_PYTHON:
      # special case for EmbeddedWorkerHandler: there's no need for a gRPC
      # server, but we need to pass self instead.  Cast to make the type check
      # on WorkerHandler.create() think we have a GrpcServer instance.
      grpc_server = cast(GrpcServer, self)
    elif self._grpc_server is None:
      self._grpc_server = GrpcServer(
          self.state_servicer, self._job_provision_info, self)
      grpc_server = self._grpc_server

    worker_handler_list = self._cached_handlers[environment_id]
    if len(worker_handler_list) < num_workers:
      for _ in range(len(worker_handler_list), num_workers):
        worker_handler = WorkerHandler.create(
            environment,
            self.state_servicer,
            self._job_provision_info.for_environment(environment),
            grpc_server)
        _LOGGER.info(
            "Created Worker handler %s for environment %s",
            worker_handler,
            environment)
        self._cached_handlers[environment_id].append(worker_handler)
        self._workers_by_id[worker_handler.worker_id] = worker_handler
        worker_handler.start_worker()
    return self._cached_handlers[environment_id][:num_workers]

  def close_all(self):
    for worker_handler_list in self._cached_handlers.values():
      for worker_handler in set(worker_handler_list):
        try:
          worker_handler.close()
        except Exception:
          _LOGGER.error(
              "Error closing worker_handler %s" % worker_handler, exc_info=True)
    self._cached_handlers = {}
    self._workers_by_id = {}
    if self._grpc_server is not None:
      self._grpc_server.close()
      self._grpc_server = None

  def get_worker(self, worker_id):
    return self._workers_by_id[worker_id]

class StateServicer(beam_fn_api_pb2_grpc.BeamFnStateServicer,
                    sdk_worker.StateHandler):
  class CopyOnWriteState(object):
    def __init__(self, underlying):
      # type: (DefaultDict[bytes, Buffer]) -> None
      self._underlying = underlying
      self._overlay = {}  # type: Dict[bytes, Buffer]

    def __getitem__(self, key):
      # type: (bytes) -> Buffer
      if key in self._overlay:
        return self._overlay[key]
      else:
        return StateServicer.CopyOnWriteList(
            self._underlying, self._overlay, key)

    def __delitem__(self, key):
      # type: (bytes) -> None
      self._overlay[key] = []

    def commit(self):
      # type: () -> DefaultDict[bytes, Buffer]
      self._underlying.update(self._overlay)
      return self._underlying

  class CopyOnWriteList(object):
    def __init__(self,
        underlying,  # type: DefaultDict[bytes, Buffer]
        overlay,  # type: Dict[bytes, Buffer]
        key  # type: bytes
    ):
      # type: (...) -> None
      self._underlying = underlying
      self._overlay = overlay
      self._key = key

    def __iter__(self):
      # type: () -> Iterator[bytes]
      if self._key in self._overlay:
        return iter(self._overlay[self._key])
      else:
        return iter(self._underlying[self._key])

    def append(self, item):
      # type: (bytes) -> None
      if self._key not in self._overlay:
        self._overlay[self._key] = list(self._underlying[self._key])
      self._overlay[self._key].append(item)

  StateType = Union[CopyOnWriteState, DefaultDict[bytes, Buffer]]

  def __init__(self):
    # type: () -> None
    self._lock = threading.Lock()
    self._state = collections.defaultdict(list)  # type: StateServicer.StateType
    self._checkpoint = None  # type: Optional[StateServicer.StateType]
    self._use_continuation_tokens = False
    self._continuations = {}  # type: Dict[bytes, Tuple[bytes, ...]]

  def checkpoint(self):
    # type: () -> None
    assert self._checkpoint is None and not \
      isinstance(self._state, StateServicer.CopyOnWriteState)
    self._checkpoint = self._state
    self._state = StateServicer.CopyOnWriteState(self._state)

  def commit(self):
    # type: () -> None
    assert isinstance(self._state,
                      StateServicer.CopyOnWriteState) and \
           isinstance(self._checkpoint,
                      StateServicer.CopyOnWriteState)
    self._state.commit()
    self._state = self._checkpoint.commit()
    self._checkpoint = None

  def restore(self):
    # type: () -> None
    assert self._checkpoint is not None
    self._state = self._checkpoint
    self._checkpoint = None

  @contextlib.contextmanager
  def process_instruction_id(self, unused_instruction_id):
    yield

  def get_raw(self,
      state_key,  # type: beam_fn_api_pb2.StateKey
      continuation_token=None  # type: Optional[bytes]
              ):
    # type: (...) -> Tuple[bytes, Optional[bytes]]
    with self._lock:
      full_state = self._state[self._to_key(state_key)]
      if self._use_continuation_tokens:
        # The token is "nonce:index".
        if not continuation_token:
          token_base = b'token_%x' % len(self._continuations)
          self._continuations[token_base] = tuple(full_state)
          return b'', b'%s:0' % token_base
        else:
          token_base, index = continuation_token.split(b':')
          ix = int(index)
          full_state_cont = self._continuations[token_base]
          if ix == len(full_state_cont):
            return b'', None
          else:
            return full_state_cont[ix], b'%s:%d' % (token_base, ix + 1)
      else:
        assert not continuation_token
        return b''.join(full_state), None

  def append_raw(
      self,
      state_key,  # type: beam_fn_api_pb2.StateKey
      data  # type: bytes
  ):
    # type: (...) -> _Future
    with self._lock:
      self._state[self._to_key(state_key)].append(data)
    return _Future.done()

  def clear(self, state_key):
    # type: (beam_fn_api_pb2.StateKey) -> _Future
    with self._lock:
      try:
        del self._state[self._to_key(state_key)]
      except KeyError:
        # This may happen with the caching layer across bundles. Caching may
        # skip this storage layer for a blocking_get(key) request. Without
        # the caching, the state for a key would be initialized via the
        # defaultdict that _state uses.
        pass
    return _Future.done()

  @staticmethod
  def _to_key(state_key):
    # type: (beam_fn_api_pb2.StateKey) -> bytes
    return state_key.SerializeToString()

class GrpcStateServicer(beam_fn_api_pb2_grpc.BeamFnStateServicer):
  def __init__(self, state):
    # type: (StateServicer) -> None
    self._state = state

  def State(self,
      request_stream,  # type: Iterable[beam_fn_api_pb2.StateRequest]
      context=None
            ):
    # type: (...) -> Iterator[beam_fn_api_pb2.StateResponse]
    # Note that this eagerly mutates state, assuming any failures are fatal.
    # Thus it is safe to ignore instruction_id.
    for request in request_stream:
      request_type = request.WhichOneof('request')
      if request_type == 'get':
        data, continuation_token = self._state.get_raw(
            request.state_key, request.get.continuation_token)
        yield beam_fn_api_pb2.StateResponse(
            id=request.id,
            get=beam_fn_api_pb2.StateGetResponse(
                data=data, continuation_token=continuation_token))
      elif request_type == 'append':
        self._state.append_raw(request.state_key, request.append.data)
        yield beam_fn_api_pb2.StateResponse(
            id=request.id, append=beam_fn_api_pb2.StateAppendResponse())
      elif request_type == 'clear':
        self._state.clear(request.state_key)
        yield beam_fn_api_pb2.StateResponse(
            id=request.id, clear=beam_fn_api_pb2.StateClearResponse())
      else:
        raise NotImplementedError('Unknown state request: %s' % request_type)

class SingletonStateHandlerFactory(sdk_worker.StateHandlerFactory):
  """A singleton cache for a StateServicer."""
  def __init__(self, state_handler):
    # type: (sdk_worker.CachingStateHandler) -> None
    self._state_handler = state_handler

  def create_state_handler(self, api_service_descriptor):
    # type: (endpoints_pb2.ApiServiceDescriptor) -> sdk_worker.CachingStateHandler

    """Returns the singleton state handler."""
    return self._state_handler

  def close(self):
    # type: () -> None

    """Does nothing."""
    pass

class ControlFuture(object):
  def __init__(self, instruction_id, response=None):
    self.instruction_id = instruction_id
    if response:
      self._response = response
    else:
      self._response = None
      self._condition = threading.Condition()
    self._exception = None

  def is_done(self):
    return self._response is not None

  def set(self, response):
    with self._condition:
      self._response = response
      self._condition.notify_all()

  def get(self, timeout=None):
    if not self._response and not self._exception:
      with self._condition:
        if not self._response and not self._exception:
          self._condition.wait(timeout)
    if self._exception:
      raise self._exception
    else:
      return self._response

  def abort(self, exception):
    with self._condition:
      self._exception = exception
      self._condition.notify_all()


text:
title: "Configuring Dependencies, Connectors, Libraries"
nav-parent_id: projectsetup
nav-pos: 2
---

Every Flink application depends on a set of Flink libraries. At the bare minimum, the application depends
on the Flink APIs. Many applications depend in addition on certain connector libraries (like Kafka, Cassandra, etc.).
When running Flink applications (either in a distributed deployment, or in the IDE for testing), the Flink
runtime library must be available as well.

## Flink Core and Application Dependencies

As with most systems that run user-defined applications, there are two broad categories of dependencies and libraries in Flink:

  - **Flink Core Dependencies**: Flink itself consists of a set of classes and dependencies that are needed to run the system, for example
    coordination, networking, checkpoints, failover, APIs, operations (such as windowing), resource management, etc.
    The set of all these classes and dependencies forms the core of Flink's runtime and must be present when a Flink
    application is started.

    These core classes and dependencies are packaged in the `flink-dist` jar. They are part of Flink's `lib` folder and
    part of the basic Flink container images. Think of these dependencies as similar to Java's core library (`rt.jar`, `charsets.jar`, etc.),
    which contains the classes like `String` and `List`.

    The Flink Core Dependencies do not contain any connectors or libraries (CEP, SQL, ML, etc.) in order to avoid having an excessive
    number of dependencies and classes in the classpath by default. In fact, we try to keep the core dependencies as slim as possible
    to keep the default classpath small and avoid dependency clashes.

  - The **User Application Dependencies** are all connectors, formats, or libraries that a specific user application needs.

    The user application is typically packaged into an *application jar*, which contains the application code and the required
    connector and library dependencies.

    The user application dependencies explicitly do not include the Flink DataSet / DataStream APIs and runtime dependencies,
    because those are already part of Flink's Core Dependencies.

## Setting up a Project: Basic Dependencies

Every Flink application needs as the bare minimum the API dependencies, to develop against.
For Maven, you can use the [Java Project Template]({{ site.baseurl }}/dev/projectsetup/java_api_quickstart.html)
or [Scala Project Template]({{ site.baseurl }}/dev/projectsetup/scala_api_quickstart.html) to create
a program skeleton with these initial dependencies.

When setting up a project manually, you need to add the following dependencies for the Java/Scala API
(here presented in Maven syntax, but the same dependencies apply to other build tools (Gradle, SBT, etc.) as well.

{% highlight xml %}
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-java</artifactId>
  <version>{{site.version }}</version>
  <scope>provided</scope>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-java{{ site.scala_version_suffix }}</artifactId>
  <version>{{site.version }}</version>
  <scope>provided</scope>
</dependency>
{% endhighlight %}

{% highlight xml %}
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-scala{{ site.scala_version_suffix }}</artifactId>
  <version>{{site.version }}</version>
  <scope>provided</scope>
</dependency>
<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-streaming-scala{{ site.scala_version_suffix }}</artifactId>
  <version>{{site.version }}</version>
  <scope>provided</scope>
</dependency>
{% endhighlight %}

**Important:** Please note that all these dependencies have their scope set to *provided*.
That means that they are needed to compile against, but that they should not be packaged into the
project's resulting application jar file - these dependencies are Flink Core Dependencies,
which are already available in any setup.

It is highly recommended to keep the dependencies in scope *provided*. If they are not set to *provided*,
the best case is that the resulting JAR becomes excessively large, because it also contains all Flink core
dependencies. The worst case is that the Flink core dependencies that are added to the application's jar file
clash with some of your own dependency versions (which is normally avoided through inverted classloading).

**Note on IntelliJ:** To make the applications run within IntelliJ IDEA, the Flink dependencies need
to be declared in scope *compile* rather than *provided*. Otherwise IntelliJ will not add them to the classpath and
the in-IDE execution will fail with a `NoClassDefFountError`. To avoid having to declare the
dependency scope as *compile* (which is not recommended, see above), the above linked Java- and Scala
project templates use a trick: They add a profile that selectively activates when the application
is run in IntelliJ and only then promotes the dependencies to scope *compile*, without affecting
the packaging of the JAR files.

## Adding Connector and Library Dependencies

Most applications need specific connectors or libraries to run, for example a connector to Kafka, Cassandra, etc.
These connectors are not part of Flink's core dependencies and must hence be added as dependencies to the application

Below is an example adding the connector for Kafka 0.10 as a dependency (Maven syntax):
{% highlight xml %}
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka-0.10{{ site.scala_version_suffix }}</artifactId>
    <version>{{site.version }}</version>
</dependency>
{% endhighlight %}

We recommend to package the application code and all its required dependencies into one *jar-with-dependencies* which
we refer to as the *application jar*. The application jar can be submitted to an already running Flink cluster,
or added to a Flink application container image.

Projects created from the [Java Project Template]({{ site.baseurl }}/dev/projectsetup/java_api_quickstart.html) or
[Scala Project Template]({{ site.baseurl }}/dev/projectsetup/scala_api_quickstart.html) are configured to automatically include
the application dependencies into the application jar when running `mvn clean package`. For projects that are
not set up from those templates, we recommend to add the Maven Shade Plugin (as listed in the Appendix below)
to build the application jar with all required dependencies.

**Important:** For Maven (and other build tools) to correctly package the dependencies into the application jar,
these application dependencies must be specified in scope *compile* (unlike the core dependencies, which
must be specified in scope *provided*).

## Scala Versions

Scala versions (2.10, 2.11, 2.12, etc.) are not binary compatible with one another.
For that reason, Flink for Scala 2.11 cannot be used with an application that uses
Scala 2.12.

All Flink dependencies that (transitively) depend on Scala are suffixed with the
Scala version that they are built for, for example `flink-streaming-scala_2.11`.

Developers that only use Java can pick any Scala version, Scala developers need to
pick the Scala version that matches their application's Scala version.

Please refer to the [build guide]({{ site.baseurl }}/flinkDev/building.html#scala-versions)
for details on how to build Flink for a specific Scala version.

## Hadoop Dependencies

**General rule: It should never be necessary to add Hadoop dependencies directly to your application.**
*(The only exception being when using existing Hadoop input-/output formats with Flink's Hadoop compatibility wrappers)*

If you want to use Flink with Hadoop, you need to have a Flink setup that includes the Hadoop dependencies, rather than
adding Hadoop as an application dependency. Please refer to the [Hadoop Setup Guide]({{ site.baseurl }}/ops/deployment/hadoop.html)
for details.

There are two main reasons for that design:

  - Some Hadoop interaction happens in Flink's core, possibly before the user application is started, for example
    setting up HDFS for checkpoints, authenticating via Hadoop's Kerberos tokens, or deployment on YARN.

  - Flink's inverted classloading approach hides many transitive dependencies from the core dependencies. That applies not only
    to Flink's own core dependencies, but also to Hadoop's dependencies when present in the setup.
    That way, applications can use different versions of the same dependencies without running into dependency conflicts (and
    trust us, that's a big deal, because Hadoops dependency tree is huge.)

If you need Hadoop dependencies during testing or development inside the IDE (for example for HDFS access), please configure
these dependencies similar to the scope of the dependencies to *test* or to *provided*.

## Appendix: Template for building a Jar with Dependencies

To build an application JAR that contains all dependencies required for declared connectors and libraries,
you can use the following shade plugin definition:

{% highlight xml %}
<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-shade-plugin</artifactId>
            <version>3.1.1</version>
            <executions>
                <execution>
                    <phase>package</phase>
                    <goals>
                        <goal>shade</goal>
                    </goals>
                    <configuration>
                        <artifactSet>
                            <excludes>
                                <exclude>com.google.code.findbugs:jsr305</exclude>
                                <exclude>org.slf4j:*</exclude>
                                <exclude>log4j:*</exclude>
                            </excludes>
                        </artifactSet>
                        <filters>
                            <filter>

                                <artifact>*:*</artifact>
                                <excludes>
                                    <exclude>META-INF/*.SF</exclude>
                                    <exclude>META-INF/*.DSA</exclude>
                                    <exclude>META-INF/*.RSA</exclude>
                                </excludes>
                            </filter>
                        </filters>
                        <transformers>
                            <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                <mainClass>my.programs.main.clazz</mainClass>
                            </transformer>
                        </transformers>
                    </configuration>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>
{% endhighlight %}

{% top %}


text:
import pbr.version

__version__ = pbr.version.VersionInfo('python-tripleoclient')


text:
namespace ash {
namespace file_system_provider {

BufferingFileStreamReader::BufferingFileStreamReader(
    std::unique_ptr<storage::FileStreamReader> file_stream_reader,
    int preloading_buffer_length,
    int64_t max_bytes_to_read)
    : file_stream_reader_(std::move(file_stream_reader)),
      preloading_buffer_length_(preloading_buffer_length),
      max_bytes_to_read_(max_bytes_to_read),
      bytes_read_(0),
      preloading_buffer_(
          base::MakeRefCounted<net::IOBuffer>(preloading_buffer_length)),
      preloading_buffer_offset_(0),
      preloaded_bytes_(0) {}

BufferingFileStreamReader::~BufferingFileStreamReader() {
}

int BufferingFileStreamReader::Read(net::IOBuffer* buffer,
                                    int buffer_length,
                                    net::CompletionOnceCallback callback) {
  // Return as much as available in the internal buffer. It may be less than
  // |buffer_length|, what is valid.
  const int bytes_read =
      CopyFromPreloadingBuffer(base::WrapRefCounted(buffer), buffer_length);
  if (bytes_read)
    return bytes_read;

  // If the internal buffer is empty, and more bytes than the internal buffer
  // size is requested, then call the internal file stream reader directly.
  if (buffer_length >= preloading_buffer_length_) {
    const int result = file_stream_reader_->Read(
        buffer, buffer_length,
        base::BindOnce(&BufferingFileStreamReader::OnReadCompleted,
                       weak_ptr_factory_.GetWeakPtr(), std::move(callback)));
    DCHECK_EQ(result, net::ERR_IO_PENDING);
    return result;
  }

  // Nothing copied, so contents have to be preloaded.
  Preload(base::BindOnce(
      &BufferingFileStreamReader::OnReadCompleted,
      weak_ptr_factory_.GetWeakPtr(),
      base::BindOnce(&BufferingFileStreamReader::OnPreloadCompleted,
                     weak_ptr_factory_.GetWeakPtr(),
                     base::WrapRefCounted(buffer), buffer_length,
                     std::move(callback))));

  return net::ERR_IO_PENDING;
}

int64_t BufferingFileStreamReader::GetLength(
    net::Int64CompletionOnceCallback callback) {
  const int64_t result = file_stream_reader_->GetLength(std::move(callback));
  DCHECK_EQ(net::ERR_IO_PENDING, result);

  return result;
}

int BufferingFileStreamReader::CopyFromPreloadingBuffer(
    scoped_refptr<net::IOBuffer> buffer,
    int buffer_length) {
  const int read_bytes = std::min(buffer_length, preloaded_bytes_);

  memcpy(buffer->data(),
         preloading_buffer_->data() + preloading_buffer_offset_,
         read_bytes);
  preloading_buffer_offset_ += read_bytes;
  preloaded_bytes_ -= read_bytes;

  return read_bytes;
}

void BufferingFileStreamReader::Preload(net::CompletionOnceCallback callback) {
  const int preload_bytes =
      std::min(static_cast<int64_t>(preloading_buffer_length_),
               max_bytes_to_read_ - bytes_read_);

  const int result = file_stream_reader_->Read(
      preloading_buffer_.get(), preload_bytes, std::move(callback));
  DCHECK_EQ(result, net::ERR_IO_PENDING);
}

void BufferingFileStreamReader::OnPreloadCompleted(
    scoped_refptr<net::IOBuffer> buffer,
    int buffer_length,
    net::CompletionOnceCallback callback,
    int result) {
  if (result < 0) {
    std::move(callback).Run(result);
    return;
  }

  preloading_buffer_offset_ = 0;
  preloaded_bytes_ = result;

  std::move(callback).Run(CopyFromPreloadingBuffer(buffer, buffer_length));
}

void BufferingFileStreamReader::OnReadCompleted(
    net::CompletionOnceCallback callback,
    int result) {
  if (result < 0) {
    std::move(callback).Run(result);
    return;
  }

  // If more bytes than declared in |max_bytes_to_read_| was read in total, then
  // emit an
  // error.
  if (result > max_bytes_to_read_ - bytes_read_) {
    std::move(callback).Run(net::ERR_FAILED);
    return;
  }

  bytes_read_ += result;
  DCHECK_LE(bytes_read_, max_bytes_to_read_);

  std::move(callback).Run(result);
}

}  // namespace file_system_provider
}  // namespace ash


text:
package de.vandermeer.skb.interfaces.application;

/**
 * A standard simple CLI option.
 *
 * @author     Sven van der Meer <vdmeer.sven@mykolab.com>
 * @version    v0.0.2 build 170502 (02-May-17) for Java 1.8
 * @since      v0.0.2
 */
public interface Apo_SimpleC extends ApoBaseC {

	/**
	 * Tests if the option is set.
	 * A simple option with CLI is set if it is present in a parsed command line.
	 * @return true if set, false otherwise
	 */
	@Override
	default boolean isSet(){
		return this.inCli();
	}

}


text:
- Maven 3+. See https://maven.apache.org/
- Docker

### Setup MongoDB for a dev environment

1. From the base directory, build the Mongo docker image via `docker build -t xbdd_mongo_dev mongo`
1. Start the docker container with `docker run -p=27017:27017 --name xbdd_mongo_dev -d xbdd_mongo_dev`

This will give you a docker container named `xbdd_mongo_dev` which is accessible at
http://localhost:27017

### Registering a Github OAuth App

1. Go to https://github.com/settings/developers and register a new OAuth application
1. Call the `application xbdd-{your-github-user-id}`
1. For the homepage url, use `http://localhost:3000`
1. For the redirect url, use `http://localhost:3000/login/oauth2/code/github`
1. In `frontend/.env.development`, add `REACT_APP_GITHUB_CLIENT_ID={your-app-client-id}`.
1. In `backend/src/main/resources/application-dev.yml` paste in your app's client id and secret into `github.client-id` and `github.client-secret` respectively.

### OPTIONAL - Setup SSL Certificates

It is possible to run the dev environment using https. If you need to do so, in `frontend/.env.development` set `REACT_APP_BACKEND_HOST=https://localhost:8443` and add the following block,

```
HTTPS=true
```

Then you need to enable the backend server to run SSL. To do that:

1. Enable the block beginning with `server` in the `application-dev.yml`;

Finally you need to set up certificates, to do so follow the instructions for both the frontend and backend below.

1. Follow the instructions in [backend/src/main/resources/certs/README.md](../backend/src/main/resources/certs/README.md)

### Running XBDD

#### Via Eclipse

1. Import the top level into Eclipse. Ensure the nested projects import as well.
1. Install Eclipse Spring Tools 4, this can be obtained through the the Eclipse Marketplace or downloaded [here](https://spring.io/tools).
1. Run `mvn clean install` in the top level directory
1. There is a Spring Boot Run Configuration called `Xbdd Run` that starts the backend.
1. In the `frontend` directory run `npm start`
1. Xbdd will be available at http://localhost:3000

#### Via the commandline

1. In the top level directory, run `mvn clean install`.
1. In `backend`, run `mvn spring-boot:run -Dspring-boot.run.profiles=dev`
1. In `frontend`, run `npm start`
1. Ensure your mongo environment is running via `http://localhost:27017`. If not, see the earlier step.
1. Xbdd will be available at `http://localhost:3000`

### Testing the Docker deploy version

This builds the docker compose deployment without mounting the mongo db to a volume, as we require a /opt/xbdd/volumes/ to exist. The only caveat is that data will not saved between deployments, to save data the volume will need to be mounted locally.
It is supposed to be used for testing changes to the deployment process.

To run do

1. `docker-compose -f docker-compose-dev.yml build`
2. `docker-compose -f docker-compose-dev.yml -d up`

To stop, `docker-compose -f docker-compose-dev.yml --remove-orphans down`