package com.github.sesameloader;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

import org.openrdf.model.Statement;
import org.openrdf.repository.RepositoryConnection;
import org.openrdf.repository.RepositoryException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class StatementFromQueueIntoRepositoryPusher
    implements Runnable
{

	private final BlockingQueue<Statement> queue;

	private final RepositoryConnection connection;
	private final int commitEveryStatements;
	private volatile boolean finished;

	private Logger log = LoggerFactory.getLogger(StatementFromQueueIntoRepositoryPusher.class);

	public StatementFromQueueIntoRepositoryPusher(BlockingQueue<Statement> queue, int commitEveryStatements,
	    RepositoryManager connection) throws RepositoryException
	{
		super();
		this.queue = queue;
		this.commitEveryStatements = commitEveryStatements;
		this.connection = connection.getConnection();
	}

	@Override
	public void run()
	{
		int counter = 0;
		if (log.isDebugEnabled())
			log.debug("Running into repository pusher");
		try
		{
			while (!this.finished || !queue.isEmpty())
				counter = takeStatementFromQueueAddToConnection(counter);
			connection.commit();
                        connection.close();
		} catch (RepositoryException e)
		{
			log.error("Pusher failed " + e.getMessage());
		}
	}

	private int takeStatementFromQueueAddToConnection(int counter)
	    throws RepositoryException
	{
		{
			try
			{
				final Statement st = queue.poll(100, TimeUnit.MILLISECONDS);
				if (st != null)
				{
					connection.add(st);
					counter++;
					if (counter % commitEveryStatements == 0)
					{
						if (log.isDebugEnabled())
							log.debug("Commiting into the connection pusher");
						connection.commit();
					}

				}
			} catch (InterruptedException e1)
			{
				Thread.currentThread().interrupt();
			}
		}
		return counter;
	}

    public void setFinished(boolean b)
    {
        finished =true;
    }
}
@@ -784,26 +784,23 @@
 Manager 
-connection
+manager
 ) throws
@@ -930,26 +930,23 @@
 ction = 
-connection
+manager
 .getConn
@@ -1233,52 +1233,8 @@
 ();%0A
-                        connection.close();%0A
 %09%09%7D 
@@ -1320,16 +1320,255 @@
 ));%0A%09%09%7D%0A
+%09%09finally%0A%09%09%7B%0A            try%0A            %7B%0A                connection.close();%0A            %7D%0A            catch(RepositoryException e)%0A            %7B%0A                log.error(%22Error closing connection to repository%22, e);%0A            %7D%0A%09%09%7D%0A
 %09%7D%0A%0A%09pri

put connection.close in a finally block to attempt to make sure it is always called