<commit_before>/*
 * (c) Copyright 2004, 2005 Hewlett-Packard Development Company, LP
 * [See end of file]
 */

package com.hp.hpl.jena.rdf.arp;


import org.xml.sax.*;

import EDU.oswego.cs.dl.util.concurrent.BoundedBuffer;

import com.hp.hpl.jena.shared.JenaException;

/**
 * @author <NAME>
 *  
 */
class PushMePullYouPipe extends TokenPipe {

	volatile private Throwable brokenPipe = null;

	private boolean open = true;
	
	private boolean overflow = false;

	private BoundedBuffer buffer = new BoundedBuffer(100);

	private Object pending = null;

	private static final String finished = "<finished>";

	private static final RuntimeException naturalEnd = new RuntimeException();

	final private Thread puller;

	
	PushMePullYouPipe(final ARPRunnable puller) {
		this.puller =  
				new Thread() {
			public void run() {
				//		pipe.pullerSleep();
				try {
					puller.run();
					naturalDeath();
				} catch (Throwable e) {
					setException(e);
				} finally {
				}
			}

		};
		
	}
    void start() {
	   this.puller.start();
    }
	/**
	 * get something from the pipe; take care of BoundedBuffer's checked
	 * exceptions
	 */
	private Object fetch() {
		try {
			return buffer.take();
		} catch (Exception e) {
			throw new BoundedBufferTakeException(e);
		}
	}

	/**
	 * put something into the pipe; take care of BoundedBuffer's checked
	 * exceptions
	 */
	private void putAny(Object d) throws SAXParseException {
		try {
			do {
				if (d != finished)
					isPipeBroken();
			} while (!buffer.offer(d, 100));
		} catch (InterruptedException e) {
			throw new BoundedBufferPutException(e);
		}
	}

	private void isPipeBroken() throws SAXParseException {
		if (brokenPipe != null) {
			if ( brokenPipe == naturalEnd ){
				SAXParseException ee = new SAXParseException("RDF parsing finished, additional XML events",getLocator());
				//System.err.println(
				//		ParseException.formatMessage(ee));
				overflow = true;
				throw ee;
			}
			try {
				throw brokenPipe;
			} catch (RuntimeException e) {
				throw e;
			} catch (Error e) {
				throw e;
			} catch (SAXParseException e) {
				throw e;
			} catch (Exception e) {
				throw new WrappedException(e);

			} catch (Throwable t) {
				throw new RuntimeException("Exception from RDF thread.",t);
			}
			/*
			pipe().naturalDeath();
		} catch (WrappedException wrapped) {
			pipe().setException(wrapped.inner);
		} catch (ParseException parse) {
			pipe().setException(parse.rootCause());
		} catch (RuntimeException e) {
			pipe().setException(e);
		} catch (Error e) {
			pipe().setException(e);
		*/
		}
	}

	public void putNextToken(Token d) throws SAXParseException {
		putAny(d);
	}

	public void close() throws SAXParseException {
		putAny(finished);
		try {
			puller.join();
		} catch (InterruptedException e) {

		}
		if (brokenPipe != naturalEnd)
			isPipeBroken();
		/* Note check that pipe is exhausted is done
		 * on exactlyExhausted ....
		 */
	}

	private boolean hasNext() {
		if (open) {
			if (pending == null) {
				pending = fetch();
				if (pending == finished)
					open = false;
				return open;
			} else
				return true;
		} else
			return false;
	}
	
	boolean exactlyExhausted() {
		return !(overflow||hasNext());
	}

	public Token getNextToken() {
		if (hasNext() == false)
			return new Token(RDFParserConstants.EOF, null);
		try {
			return (Token) pending;
		} finally {
			pending = null;
		}
	}

	void naturalDeath() {
		setException(naturalEnd);
	}

	void setException(Throwable t) {
		brokenPipe = t;
	}

	/**
	 * Exception to throw if a <code>take</code> throws an exception.
	 */
	public static class BoundedBufferTakeException extends JenaException {
		BoundedBufferTakeException(Exception e) {
			super(e);
		}
	}

	/**
	 * Exception to throw if a <code>put</code> throws an exception.
	 */
	public static class BoundedBufferPutException extends JenaException {
		BoundedBufferPutException(Exception e) {
			super(e);<commit_msg>made exception classes package scope instead of public<commit_after>/*
 * (c) Copyright 2004, 2005 Hewlett-Packard Development Company, LP
 * [See end of file]
 */

package com.hp.hpl.jena.rdf.arp;


import org.xml.sax.*;

import EDU.oswego.cs.dl.util.concurrent.BoundedBuffer;

import com.hp.hpl.jena.shared.JenaException;

/**
 * @author <NAME>
 *  
 */
class PushMePullYouPipe extends TokenPipe {

	volatile private Throwable brokenPipe = null;

	private boolean open = true;
	
	private boolean overflow = false;

	private BoundedBuffer buffer = new BoundedBuffer(100);

	private Object pending = null;

	private static final String finished = "<finished>";

	private static final RuntimeException naturalEnd = new RuntimeException();

	final private Thread puller;

	
	PushMePullYouPipe(final ARPRunnable puller) {
		this.puller =  
				new Thread() {
			public void run() {
				//		pipe.pullerSleep();
				try {
					puller.run();
					naturalDeath();
				} catch (Throwable e) {
					setException(e);
				} finally {
				}
			}

		};
		
	}
    void start() {
	   this.puller.start();
    }
	/**
	 * get something from the pipe; take care of BoundedBuffer's checked
	 * exceptions
	 */
	private Object fetch() {
		try {
			return buffer.take();
		} catch (Exception e) {
			throw new BoundedBufferTakeException(e);
		}
	}

	/**
	 * put something into the pipe; take care of BoundedBuffer's checked
	 * exceptions
	 */
	private void putAny(Object d) throws SAXParseException {
		try {
			do {
				if (d != finished)
					isPipeBroken();
			} while (!buffer.offer(d, 100));
		} catch (InterruptedException e) {
			throw new BoundedBufferPutException(e);
		}
	}

	private void isPipeBroken() throws SAXParseException {
		if (brokenPipe != null) {
			if ( brokenPipe == naturalEnd ){
				SAXParseException ee = new SAXParseException("RDF parsing finished, additional XML events",getLocator());
				//System.err.println(
				//		ParseException.formatMessage(ee));
				overflow = true;
				throw ee;
			}
			try {
				throw brokenPipe;
			} catch (RuntimeException e) {
				throw e;
			} catch (Error e) {
				throw e;
			} catch (SAXParseException e) {
				throw e;
			} catch (Exception e) {
				throw new WrappedException(e);

			} catch (Throwable t) {
				throw new RuntimeException("Exception from RDF thread.",t);
			}
			/*
			pipe().naturalDeath();
		} catch (WrappedException wrapped) {
			pipe().setException(wrapped.inner);
		} catch (ParseException parse) {
			pipe().setException(parse.rootCause());
		} catch (RuntimeException e) {
			pipe().setException(e);
		} catch (Error e) {
			pipe().setException(e);
		*/
		}
	}

	public void putNextToken(Token d) throws SAXParseException {
		putAny(d);
	}

	public void close() throws SAXParseException {
		putAny(finished);
		try {
			puller.join();
		} catch (InterruptedException e) {

		}
		if (brokenPipe != naturalEnd)
			isPipeBroken();
		/* Note check that pipe is exhausted is done
		 * on exactlyExhausted ....
		 */
	}

	private boolean hasNext() {
		if (open) {
			if (pending == null) {
				pending = fetch();
				if (pending == finished)
					open = false;
				return open;
			} else
				return true;
		} else
			return false;
	}
	
	boolean exactlyExhausted() {
		return !(overflow||hasNext());
	}

	public Token getNextToken() {
		if (hasNext() == false)
			return new Token(RDFParserConstants.EOF, null);
		try {
			return (Token) pending;
		} finally {
			pending = null;
		}
	}

	void naturalDeath() {
		setException(naturalEnd);
	}

	void setException(Throwable t) {
		brokenPipe = t;
	}

	/**
	 * Exception to throw if a <code>take</code> throws an exception.
	 */
	static class BoundedBufferTakeException extends JenaException {
		BoundedBufferTakeException(Exception e) {
			super(e);
		}
	}

	/**
	 * Exception to throw if a <code>put</code> throws an exception.
	 */
	static class BoundedBufferPutException extends JenaException {
		BoundedBufferPutException(Exception e) {
			super(e);
{'fraction_non_alphanumeric': 0.11989376501833818, 'fraction_numerical': 0.0035411660553939547, 'mean_word_length': 3.170886075949367, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '3308402', 'n_tokens_mistral': 3036, 'n_tokens_neox': 2503, 'n_words': 680}