<commit_before>package com.marklogic.semantics.sesame;

// https://github.com/marklogic/marklogic-sesame/issues/282
//

import static java.util.Arrays.asList;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.junit.After;
import org.junit.Test;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.openrdf.repository.RepositoryException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.marklogic.client.DatabaseClientFactory;
import com.marklogic.client.semantics.GraphManager;

public class MultiThreadedPersistenceTest extends SesameTestBase {

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    //create some data
    List<String> identifiers = asList(
            "subject1", "subject2", "subject3","subject4","subject5",
            "subject6","subject7","subject8","subject9","subject10",
            "subject11","subject12","subject13","subject14","subject15",
            "subject16", "subject17", "subject18","subject19","subject20",
            "subject21", "subject22", "subject23","subject24","subject25",
            "subject26", "subject27", "subject28","subject29","subject30",
            "subject31", "subject32", "subject33","subject34","subject35",
            "subject36", "subject37", "subject38","subject39","subject40");

    @After
    public void tearDown()
            throws Exception {
        GraphManager gmgr = adminClient.newGraphManager();
        gmgr.delete("http://graph/meta");
        for (String id : identifiers)
        {
            gmgr.delete("http://foo/graph/" + id);
        }
    }


    @Test
    public void multiThreadedPersist() throws RepositoryException, InterruptedException {
        final PersistenceService persistenceService = new PersistenceService(SesameTestBase.host, SesameTestBase.port, SesameTestBase.adminUser, SesameTestBase.adminPassword, DatabaseClientFactory.Authentication.DIGEST.toString());

        //persist data with multiple threads against the persistence service - simulate multiple concurrent requests against a tomcat deployed ingestion service
        //results in intermittent MarkLogicTransactionExceptions in executor threads
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        try {
            for(final String identifier: identifiers) {
                for (int i=0; i<20; i++) {
                    executorService.submit(
                            new Runnable() {
                                @Override
                                public void run() {
                                    persistenceService.persist(entitiesFor(identifier));
                                }
                            }
                    );
                }
            }

            executorService.shutdown();
            executorService.awaitTermination(60, TimeUnit.SECONDS);
        } finally {
            if(!executorService.isTerminated()) {
                System.out.println("cancel non finished tasks");
            }
            executorService.shutdownNow();
            System.out.println("shut down finished");
        }
    }

    class PersistenceService {
        private MarkLogicRepository markLogicRepository;

        public PersistenceService(String host, int port, String user, String password, String digest) {
            markLogicRepository = new MarkLogicRepository(host, port, user, password, digest);
            try {
                markLogicRepository.initialize();
            } catch (RepositoryException e) {
                throw new RuntimeException("error initialising repo", e);
            }
        }

        public void persist(List<Entity> entities){

            MarkLogicRepositoryConnection connection = null;
            try {
                connection = markLogicRepository.getConnection();
                for(Entity e : entities) {
                    connection.add(e.getStatements(), e.getGraph());
                }
            } catch (RepositoryException e) {
                //print to sysout as thread exceptions are not propagated up to main thread
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }

    }

    @Test
    public void singleConnectionMultiThreadedPersist() throws RepositoryException, InterruptedException {<commit_msg>ignore some tests until new java api client<commit_after>package com.marklogic.semantics.sesame;

// https://github.com/marklogic/marklogic-sesame/issues/282
//

import static java.util.Arrays.asList;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import org.junit.After;
import org.junit.Ignore;
import org.junit.Test;
import org.openrdf.model.Resource;
import org.openrdf.model.Statement;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.openrdf.repository.RepositoryException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.marklogic.client.DatabaseClientFactory;
import com.marklogic.client.semantics.GraphManager;

public class MultiThreadedPersistenceTest extends SesameTestBase {

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    //create some data
    List<String> identifiers = asList(
            "subject1", "subject2", "subject3","subject4","subject5",
            "subject6","subject7","subject8","subject9","subject10",
            "subject11","subject12","subject13","subject14","subject15",
            "subject16", "subject17", "subject18","subject19","subject20",
            "subject21", "subject22", "subject23","subject24","subject25",
            "subject26", "subject27", "subject28","subject29","subject30",
            "subject31", "subject32", "subject33","subject34","subject35",
            "subject36", "subject37", "subject38","subject39","subject40");

    @After
    public void tearDown()
            throws Exception {
        GraphManager gmgr = adminClient.newGraphManager();
        gmgr.delete("http://graph/meta");
        for (String id : identifiers)
        {
            gmgr.delete("http://foo/graph/" + id);
        }
    }


    // ignore until we have new java api client
    @Test
    @Ignore
    public void multiThreadedPersist() throws RepositoryException, InterruptedException {
        final PersistenceService persistenceService = new PersistenceService(SesameTestBase.host, SesameTestBase.port, SesameTestBase.adminUser, SesameTestBase.adminPassword, DatabaseClientFactory.Authentication.DIGEST.toString());

        //persist data with multiple threads against the persistence service - simulate multiple concurrent requests against a tomcat deployed ingestion service
        //results in intermittent MarkLogicTransactionExceptions in executor threads
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        try {
            for(final String identifier: identifiers) {
                for (int i=0; i<20; i++) {
                    executorService.submit(
                            new Runnable() {
                                @Override
                                public void run() {
                                    persistenceService.persist(entitiesFor(identifier));
                                }
                            }
                    );
                }
            }

            executorService.shutdown();
            executorService.awaitTermination(60, TimeUnit.SECONDS);
        } finally {
            if(!executorService.isTerminated()) {
                System.out.println("cancel non finished tasks");
            }
            executorService.shutdownNow();
            System.out.println("shut down finished");
        }
    }

    class PersistenceService {
        private MarkLogicRepository markLogicRepository;

        public PersistenceService(String host, int port, String user, String password, String digest) {
            markLogicRepository = new MarkLogicRepository(host, port, user, password, digest);
            try {
                markLogicRepository.initialize();
            } catch (RepositoryException e) {
                throw new RuntimeException("error initialising repo", e);
            }
        }

        public void persist(List<Entity> entities){

            MarkLogicRepositoryConnection connection = null;
            try {
                connection = markLogicRepository.getConnection();
                for(Entity e : entities) {
                    connection.add(e.getStatements(), e.getGraph());
                }
            } catch (RepositoryException e) {
                //print to sysout as thread exceptions are not propagated up to main thread
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }

    }

    // ignore until we have new java api client
    @Test
    @Ignore
    public void singleConnectionMultiThreadedPersist() throws RepositoryException, InterruptedException {
{'fraction_non_alphanumeric': 0.09694153247595375, 'fraction_numerical': 0.017940127526207716, 'mean_word_length': 2.3922287390029324, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3528512', 'n_tokens_mistral': 2402, 'n_tokens_neox': 2231, 'n_words': 507}