Author: kwright
Date: Sun Jan 11 16:20:23 2015
New Revision: 1650911
URL: http://svn.apache.org/r1650911
Log:
Fix for CONNECTORS-1139.
Added:
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/AbortChecker.java
(with props)
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/BreakException.java
(with props)
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IBreakCheck.java
(with props)
Modified:
manifoldcf/trunk/CHANGES.txt
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/IThrottledConnection.java
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/RSSConnector.java
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/Robots.java
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/ThrottledFetcher.java
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/IThrottledConnection.java
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/ThrottledFetcher.java
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/WebcrawlerConnector.java
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IConnectionThrottler.java
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IFetchThrottler.java
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IStreamThrottler.java
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ConnectionBin.java
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/FetchBin.java
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ThrottleBin.java
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/Throttler.java
Modified: manifoldcf/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/CHANGES.txt?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
--- manifoldcf/trunk/CHANGES.txt (original)
+++ manifoldcf/trunk/CHANGES.txt Sun Jan 11 16:20:23 2015
@@ -3,6 +3,9 @@ $Id$
======================= 2.1-dev =====================
+CONNECTORS-1139: Add support for interruptible throttlers.
+(Karl Wright)
+
CONNECTORS-1138: Loss of all metadata fields but one on pipeline
bifurcation.
(Salih Sen, Karl Wright)
Modified:
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/IThrottledConnection.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/IThrottledConnection.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/IThrottledConnection.java
(original)
+++
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/IThrottledConnection.java
Sun Jan 11 16:20:23 2015
@@ -51,7 +51,7 @@ public interface IThrottledConnection
* is used solely for logging purposes.
*/
public void beginFetch(String fetchType)
- throws ManifoldCFException;
+ throws ManifoldCFException, ServiceInterruption;
/** Execute the fetch and get the return code. This method uses the
* standard logging mechanism to keep track of the fetch attempt. It also
Modified:
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/RSSConnector.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/RSSConnector.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/RSSConnector.java
(original)
+++
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/RSSConnector.java
Sun Jan 11 16:20:23 2015
@@ -969,7 +969,8 @@ public class RSSConnector extends org.ap
proxyPort,
proxyAuthDomain,
proxyAuthUsername,
- proxyAuthPassword);
+ proxyAuthPassword,
+ activities);
try
{
// Begin the fetch
Modified:
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/Robots.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/Robots.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/Robots.java
(original)
+++
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/Robots.java
Sun Jan 11 16:20:23 2015
@@ -445,7 +445,8 @@ public class Robots
// Do the fetch
IThrottledConnection connection =
fetcher.createConnection(threadContext,throttleGroupName,
hostName,connectionLimit,ROBOT_TIMEOUT_MILLISECONDS,
-
proxyHost,proxyPort,proxyAuthDomain,proxyAuthUsername,proxyAuthPassword);
+
proxyHost,proxyPort,proxyAuthDomain,proxyAuthUsername,proxyAuthPassword,
+ activities);
try
{
connection.beginFetch(ROBOT_CONNECTION_TYPE);
Modified:
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/ThrottledFetcher.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/ThrottledFetcher.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/ThrottledFetcher.java
(original)
+++
manifoldcf/trunk/connectors/rss/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/rss/ThrottledFetcher.java
Sun Jan 11 16:20:23 2015
@@ -156,7 +156,8 @@ public class ThrottledFetcher
*/
public synchronized IThrottledConnection createConnection(IThreadContext
threadContext, String throttleGroupName,
String serverName, int connectionLimit, int connectionTimeoutMilliseconds,
- String proxyHost, int proxyPort, String proxyAuthDomain, String
proxyAuthUsername, String proxyAuthPassword)
+ String proxyHost, int proxyPort, String proxyAuthDomain, String
proxyAuthUsername, String proxyAuthPassword,
+ IAbortActivity activities)
throws ManifoldCFException, ServiceInterruption
{
IConnectionThrottler server;
@@ -171,7 +172,8 @@ public class ThrottledFetcher
return new ThrottledConnection(serverName, server,
connectionTimeoutMilliseconds,connectionLimit,
- proxyHost,proxyPort,proxyAuthDomain,proxyAuthUsername,proxyAuthPassword);
+ proxyHost,proxyPort,proxyAuthDomain,proxyAuthUsername,proxyAuthPassword,
+ activities);
}
/** Poll. This method is designed to allow idle connections to be closed
and freed.
@@ -239,18 +241,23 @@ public class ThrottledFetcher
/** Set if thread has been started */
protected boolean threadStarted = false;
+ /** Abort checker */
+ protected final AbortChecker abortChecker;
+
/** Constructor.
*/
public ThrottledConnection(String serverName,
IConnectionThrottler connectionThrottler,
int connectionTimeoutMilliseconds, int connectionLimit,
- String proxyHost, int proxyPort, String proxyAuthDomain, String
proxyAuthUsername, String proxyAuthPassword)
- throws ManifoldCFException
+ String proxyHost, int proxyPort, String proxyAuthDomain, String
proxyAuthUsername, String proxyAuthPassword,
+ IAbortActivity activities)
+ throws ManifoldCFException, ServiceInterruption
{
this.serverName = serverName;
this.connectionThrottler = connectionThrottler;
this.connectionTimeoutMilliseconds = connectionTimeoutMilliseconds;
-
+ this.abortChecker = new AbortChecker(activities);
+
// Create the https scheme for this connection
javax.net.ssl.SSLSocketFactory httpsSocketFactory =
KeystoreManagerFactory.getTrustingSecureSocketFactory();;
SSLConnectionSocketFactory myFactory = new
SSLConnectionSocketFactory(new
InterruptibleSocketFactory(httpsSocketFactory,connectionTimeoutMilliseconds),
@@ -308,35 +315,44 @@ public class ThrottledFetcher
registerGlobalHandle(connectionLimit);
try
{
- int result = connectionThrottler.waitConnectionAvailable();
+ int result = connectionThrottler.waitConnectionAvailable(abortChecker);
if (result != IConnectionThrottler.CONNECTION_FROM_CREATION)
throw new IllegalStateException("Got back unexpected value from
waitForAConnection() of "+result);
- fetchThrottler = connectionThrottler.getNewConnectionFetchThrottler();
}
catch (InterruptedException e)
{
throw new
ManifoldCFException(e.getMessage(),ManifoldCFException.INTERRUPTED);
}
+ catch (BreakException e)
+ {
+ abortChecker.rethrowExceptions();
+ }
+ fetchThrottler = connectionThrottler.getNewConnectionFetchThrottler();
}
/** Begin the fetch process.
* @param fetchType is a short descriptive string describing the kind of
fetch being requested. This
* is used solely for logging purposes.
*/
+ @Override
public void beginFetch(String fetchType)
- throws ManifoldCFException
+ throws ManifoldCFException, ServiceInterruption
{
this.fetchType = fetchType;
fetchCounter = 0L;
try
{
- if (fetchThrottler.obtainFetchDocumentPermission() == false)
+ if (fetchThrottler.obtainFetchDocumentPermission(abortChecker) ==
false)
throw new IllegalStateException("obtainFetchDocumentPermission() had
unexpected return value");
}
catch (InterruptedException e)
{
throw new
ManifoldCFException("Interrupted",ManifoldCFException.INTERRUPTED);
}
+ catch (BreakException e)
+ {
+ abortChecker.rethrowExceptions();
+ }
threadStarted = false;
}
@@ -367,6 +383,7 @@ public class ThrottledFetcher
* @param lastModified is the requested lastModified header value.
* @return the status code: success, static error, or dynamic error.
*/
+ @Override
public int executeFetch(String protocol, int port, String urlPath, String
userAgent, String from,
String lastETag, String lastModified)
throws ManifoldCFException, ServiceInterruption
@@ -534,6 +551,7 @@ public class ThrottledFetcher
/** Get the http response code.
*@return the response code. This is either an HTTP response code, or one
of the codes above.
*/
+ @Override
public int getResponseCode()
throws ManifoldCFException, ServiceInterruption
{
@@ -543,6 +561,7 @@ public class ThrottledFetcher
/** Get the response input stream. It is the responsibility of the caller
* to close this stream when done.
*/
+ @Override
public InputStream getResponseBodyStream()
throws ManifoldCFException, ServiceInterruption
{
@@ -612,6 +631,7 @@ public class ThrottledFetcher
*@param headerName is the name of the header.
*@return the header value, or null if it doesn't exist.
*/
+ @Override
public String getResponseHeader(String headerName)
throws ManifoldCFException, ServiceInterruption
{
@@ -680,6 +700,7 @@ public class ThrottledFetcher
/** Done with the fetch. Call this when the fetch has been completed. A
log entry will be generated
* describing what was done.
*/
+ @Override
public void doneFetch(IProcessActivity activities)
throws ManifoldCFException
{
@@ -730,6 +751,7 @@ public class ThrottledFetcher
/** Close the connection. Call this to end this server connection.
*/
+ @Override
public void close()
throws ManifoldCFException
{
@@ -1172,5 +1194,51 @@ public class ThrottledFetcher
}
-
+ /** This class furnishes an abort signal whenever the job activity says it
should.
+ * It should never be invoked from a background thread, only from a
ManifoldCF thread.
+ */
+ protected static class AbortChecker implements IBreakCheck
+ {
+ protected final IAbortActivity activities;
+ protected ServiceInterruption serviceInterruption = null;
+ protected ManifoldCFException mcfException = null;
+
+ public AbortChecker(IAbortActivity activities)
+ {
+ this.activities = activities;
+ }
+
+ @Override
+ public long abortCheck()
+ throws BreakException, InterruptedException
+ {
+ try
+ {
+ activities.checkJobStillActive();
+ return 1000L;
+ }
+ catch (ServiceInterruption e)
+ {
+ serviceInterruption = e;
+ throw new BreakException("Break requested: "+e.getMessage(),e);
+ }
+ catch (ManifoldCFException e)
+ {
+ if (e.getErrorCode() == ManifoldCFException.INTERRUPTED)
+ throw new InterruptedException("Interrupted: "+e.getMessage());
+ mcfException = e;
+ throw new BreakException("Error during break check:
"+e.getMessage(),e);
+ }
+ }
+
+ public void rethrowExceptions()
+ throws ManifoldCFException, ServiceInterruption
+ {
+ if (serviceInterruption != null)
+ throw serviceInterruption;
+ if (mcfException != null)
+ throw mcfException;
+ }
+ }
+
}
Added:
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/AbortChecker.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/AbortChecker.java?rev=1650911&view=auto
==============================================================================
---
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/AbortChecker.java
(added)
+++
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/AbortChecker.java
Sun Jan 11 16:20:23 2015
@@ -0,0 +1,71 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.manifoldcf.crawler.connectors.webcrawler;
+
+import org.apache.manifoldcf.core.interfaces.*;
+import org.apache.manifoldcf.connectorcommon.interfaces.*;
+import org.apache.manifoldcf.agents.interfaces.*;
+import org.apache.manifoldcf.crawler.interfaces.*;
+
+/** This class furnishes an abort signal whenever the job activity says it
should.
+* It should never be invoked from a background thread, only from a ManifoldCF
thread.
+*/
+public class AbortChecker implements IBreakCheck
+{
+ protected final IAbortActivity activities;
+ protected ServiceInterruption serviceInterruption = null;
+ protected ManifoldCFException mcfException = null;
+
+ public AbortChecker(IAbortActivity activities)
+ {
+ this.activities = activities;
+ }
+
+ @Override
+ public long abortCheck()
+ throws BreakException, InterruptedException
+ {
+ try
+ {
+ activities.checkJobStillActive();
+ return 1000L;
+ }
+ catch (ServiceInterruption e)
+ {
+ serviceInterruption = e;
+ throw new BreakException("Break requested: "+e.getMessage(),e);
+ }
+ catch (ManifoldCFException e)
+ {
+ if (e.getErrorCode() == ManifoldCFException.INTERRUPTED)
+ throw new InterruptedException("Interrupted: "+e.getMessage());
+ mcfException = e;
+ throw new BreakException("Error during break check: "+e.getMessage(),e);
+ }
+ }
+
+ public void rethrowExceptions()
+ throws ManifoldCFException, ServiceInterruption
+ {
+ if (serviceInterruption != null)
+ throw serviceInterruption;
+ if (mcfException != null)
+ throw mcfException;
+ }
+}
Propchange:
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/AbortChecker.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/AbortChecker.java
------------------------------------------------------------------------------
svn:keywords = Id
Modified:
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/IThrottledConnection.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/IThrottledConnection.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/IThrottledConnection.java
(original)
+++
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/IThrottledConnection.java
Sun Jan 11 16:20:23 2015
@@ -39,6 +39,10 @@ public interface IThrottledConnection
public static final int FETCH_INTERRUPTED = -104;
public static final int FETCH_UNKNOWN_ERROR = -999;
+ /** Set the abort checker. This must be done before the connection is
actually used.
+ */
+ public void setAbortChecker(AbortChecker abortCheck);
+
/** Check whether the connection has expired.
*@param currentTime is the current time to use to judge if a connection has
expired.
*@return true if the connection has expired, and should be closed.
@@ -50,7 +54,7 @@ public interface IThrottledConnection
* is used solely for logging purposes.
*/
public void beginFetch(String fetchType)
- throws ManifoldCFException;
+ throws ManifoldCFException, ServiceInterruption;
/** Execute the fetch and get the return code. This method uses the
* standard logging mechanism to keep track of the fetch attempt. It also
Modified:
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/ThrottledFetcher.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/ThrottledFetcher.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/ThrottledFetcher.java
(original)
+++
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/ThrottledFetcher.java
Sun Jan 11 16:20:23 2015
@@ -182,8 +182,9 @@ public class ThrottledFetcher
IKeystoreManager trustStore,
IThrottleSpec throttleDescription, String[] binNames,
int connectionLimit,
- String proxyHost, int proxyPort, String proxyAuthDomain, String
proxyAuthUsername, String proxyAuthPassword)
- throws ManifoldCFException
+ String proxyHost, int proxyPort, String proxyAuthDomain, String
proxyAuthUsername, String proxyAuthPassword,
+ IAbortActivity activities)
+ throws ManifoldCFException, ServiceInterruption
{
// Get a throttle groups handle
IThrottleGroups throttleGroups = ThrottleGroupsFactory.make(threadContext);
@@ -224,14 +225,7 @@ public class ThrottledFetcher
}
}
- try
- {
- return p.grab();
- }
- catch (InterruptedException e)
- {
- throw new
ManifoldCFException(e.getMessage(),ManifoldCFException.INTERRUPTED);
- }
+ return p.grab(activities);
}
/** Flush connections that have timed out from inactivity. */
@@ -306,7 +300,9 @@ public class ThrottledFetcher
/** Set if thread has been started */
protected boolean threadStarted = false;
-
+ /** Abort checker */
+ protected AbortChecker abortCheck = null;
+
/** Constructor. Create a connection with a specific server and port, and
* register it as active against all bins. */
public ThrottledConnection(ConnectionPool myPool, IFetchThrottler
fetchThrottler,
@@ -328,6 +324,14 @@ public class ThrottledFetcher
this.httpsSocketFactory = httpsSocketFactory;
}
+ /** Set the abort checker. This must be done before the connection is
actually used.
+ */
+ @Override
+ public void setAbortChecker(AbortChecker abortCheck)
+ {
+ this.abortCheck = abortCheck;
+ }
+
/** Check whether the connection has expired.
*@param currentTime is the current time to use to judge if a connection
has expired.
*@return true if the connection has expired, and should be closed.
@@ -369,19 +373,23 @@ public class ThrottledFetcher
*/
@Override
public void beginFetch(String fetchType)
- throws ManifoldCFException
+ throws ManifoldCFException, ServiceInterruption
{
this.fetchType = fetchType;
this.fetchCounter = 0L;
try
{
- if (fetchThrottler.obtainFetchDocumentPermission() == false)
+ if (fetchThrottler.obtainFetchDocumentPermission(abortCheck) == false)
throw new IllegalStateException("Unexpected return value from
obtainFetchDocumentPermission()");
}
catch (InterruptedException e)
{
throw new
ManifoldCFException("Interrupted",ManifoldCFException.INTERRUPTED);
}
+ catch (BreakException e)
+ {
+ abortCheck.rethrowExceptions();
+ }
}
/** Execute the fetch and get the return code. This method uses the
@@ -1960,28 +1968,44 @@ public class ThrottledFetcher
this.proxyAuthPassword = proxyAuthPassword;
}
- public IThrottledConnection grab()
- throws InterruptedException
+ public IThrottledConnection grab(IAbortActivity activities)
+ throws ManifoldCFException, ServiceInterruption
{
- // Wait for a connection
- int result = connectionThrottler.waitConnectionAvailable();
- if (result == IConnectionThrottler.CONNECTION_FROM_POOL)
+ AbortChecker abortCheck = new AbortChecker(activities);
+ try
{
- // We are guaranteed to have a connection in the pool, unless there's
a coding error.
- synchronized (connections)
+ // Wait for a connection
+ IThrottledConnection connection;
+ int result = connectionThrottler.waitConnectionAvailable(abortCheck);
+ if (result == IConnectionThrottler.CONNECTION_FROM_POOL)
+ {
+ // We are guaranteed to have a connection in the pool, unless
there's a coding error.
+ synchronized (connections)
+ {
+ connection = connections.remove(connections.size()-1);
+ }
+ }
+ else if (result == IConnectionThrottler.CONNECTION_FROM_CREATION)
{
- return connections.remove(connections.size()-1);
+ connection = new
ThrottledConnection(this,connectionThrottler.getNewConnectionFetchThrottler(),
+ protocol,server,port,authentication,baseFactory,
+ proxyHost,proxyPort,
+ proxyAuthDomain,proxyAuthUsername,proxyAuthPassword);
}
+ else
+ throw new IllegalStateException("Unexpected return value from
waitConnectionAvailable(): "+result);
+ connection.setAbortChecker(abortCheck);
+ return connection;
}
- else if (result == IConnectionThrottler.CONNECTION_FROM_CREATION)
+ catch (InterruptedException e)
{
- return new
ThrottledConnection(this,connectionThrottler.getNewConnectionFetchThrottler(),
- protocol,server,port,authentication,baseFactory,
- proxyHost,proxyPort,
- proxyAuthDomain,proxyAuthUsername,proxyAuthPassword);
+ throw new ManifoldCFException("Interrupted:
"+e.getMessage(),ManifoldCFException.INTERRUPTED);
+ }
+ catch (BreakException e)
+ {
+ abortCheck.rethrowExceptions();
+ return null;
}
- else
- throw new IllegalStateException("Unexpected return value from
waitConnectionAvailable(): "+result);
}
public void release(IThrottledConnection connection)
@@ -1995,6 +2019,7 @@ public class ThrottledFetcher
else
{
// Return to pool
+ connection.setAbortChecker(null);
synchronized (connections)
{
connections.add(connection);
Modified:
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/WebcrawlerConnector.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/WebcrawlerConnector.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/WebcrawlerConnector.java
(original)
+++
manifoldcf/trunk/connectors/webcrawler/connector/src/main/java/org/apache/manifoldcf/crawler/connectors/webcrawler/WebcrawlerConnector.java
Sun Jan 11 16:20:23 2015
@@ -855,7 +855,8 @@ public class WebcrawlerConnector extends
throttleGroupName,
protocol,ipAddress,port,
credential,trustStore,throttleDescription,binNames,connectionLimit,
-
proxyHost,proxyPort,proxyAuthDomain,proxyAuthUsername,proxyAuthPassword);
+
proxyHost,proxyPort,proxyAuthDomain,proxyAuthUsername,proxyAuthPassword,
+ activities);
try
{
connection.beginFetch((fetchStatus.sessionState ==
SESSIONSTATE_LOGIN)?FETCH_LOGIN:FETCH_STANDARD);
@@ -5343,7 +5344,8 @@ public class WebcrawlerConnector extends
IThrottledConnection connection =
ThrottledFetcher.getConnection(currentContext,throttleGroupName,
protocol,hostIPAddress,port,credential,
trustStore,throttleDescription,binNames,connectionLimit,
-
proxyHost,proxyPort,proxyAuthDomain,proxyAuthUsername,proxyAuthPassword);
+
proxyHost,proxyPort,proxyAuthDomain,proxyAuthUsername,proxyAuthPassword,
+ versionActivities);
try
{
connection.beginFetch(FETCH_ROBOTS);
Added:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/BreakException.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/BreakException.java?rev=1650911&view=auto
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/BreakException.java
(added)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/BreakException.java
Sun Jan 11 16:20:23 2015
@@ -0,0 +1,39 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.manifoldcf.connectorcommon.interfaces;
+
+/** Exception signalling that a IBreakCheck abort check caused an abort.
+*/
+public class BreakException extends Exception
+{
+ public BreakException()
+ {
+ super();
+ }
+
+ public BreakException(String msg)
+ {
+ super(msg);
+ }
+
+ public BreakException(String msg, Throwable e)
+ {
+ super(msg,e);
+ }
+}
Propchange:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/BreakException.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/BreakException.java
------------------------------------------------------------------------------
svn:keywords = Id
Added:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IBreakCheck.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IBreakCheck.java?rev=1650911&view=auto
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IBreakCheck.java
(added)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IBreakCheck.java
Sun Jan 11 16:20:23 2015
@@ -0,0 +1,32 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.manifoldcf.connectorcommon.interfaces;
+
+/** Allow for abort checks during long waits.
+*/
+public interface IBreakCheck
+{
+ /** Obtain the maximum time (in milliseconds) that a wait should occur
+ * before checking for an abort. Also check for the abort and throw a
BreakException
+ * if the abort should take place.
+ */
+ public long abortCheck()
+ throws BreakException, InterruptedException;
+
+}
Propchange:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IBreakCheck.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IBreakCheck.java
------------------------------------------------------------------------------
svn:keywords = Id
Modified:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IConnectionThrottler.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IConnectionThrottler.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IConnectionThrottler.java
(original)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IConnectionThrottler.java
Sun Jan 11 16:20:23 2015
@@ -58,7 +58,17 @@ public interface IConnectionThrottler
*/
public int waitConnectionAvailable()
throws InterruptedException;
-
+
+ /** Get permission to grab a connection for use. If this object believes
there is a connection
+ * available in the pool, it will update its pool size variable and return
If not, this method
+ * evaluates whether a new connection should be created. If neither
condition is true, it
+ * waits until a connection is available.
+ *@return whether to take the connection from the pool, or create one, or
whether the
+ * throttler is being shut down.
+ */
+ public int waitConnectionAvailable(IBreakCheck breakCheck)
+ throws InterruptedException, BreakException;
+
/** For a new connection, obtain the fetch throttler to use for the
connection.
* If the result from waitConnectionAvailable() is CONNECTION_FROM_CREATION,
* the calling code is expected to create a connection using the result of
this method.
Modified:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IFetchThrottler.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IFetchThrottler.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IFetchThrottler.java
(original)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IFetchThrottler.java
Sun Jan 11 16:20:23 2015
@@ -35,6 +35,14 @@ public interface IFetchThrottler
*/
public boolean obtainFetchDocumentPermission()
throws InterruptedException;
+
+ /** Get permission to fetch a document. This grants permission to start
+ * fetching a single document, within the connection that has already been
+ * granted permission that created this object.
+ *@return false if the throttler is being shut down.
+ */
+ public boolean obtainFetchDocumentPermission(IBreakCheck breakCheck)
+ throws InterruptedException, BreakException;
/** Open a fetch stream. When done (or aborting), call
* IStreamThrottler.closeStream() to note the completion of the document
Modified:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IStreamThrottler.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IStreamThrottler.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IStreamThrottler.java
(original)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/interfaces/IStreamThrottler.java
Sun Jan 11 16:20:23 2015
@@ -36,7 +36,16 @@ public interface IStreamThrottler
*/
public boolean obtainReadPermission(int byteCount)
throws InterruptedException;
-
+
+ /** Obtain permission to read a block of bytes. This method may wait until
it is OK to proceed.
+ * The throttle group, bin names, etc are already known
+ * to this specific interface object, so it is unnecessary to include them
here.
+ *@param byteCount is the number of bytes to get permissions to read.
+ *@return true if the wait took place as planned, or false if the system is
being shut down.
+ */
+ public boolean obtainReadPermission(int byteCount, IBreakCheck breakCheck)
+ throws InterruptedException, BreakException;
+
/** Note the completion of the read of a block of bytes. Call this after
* obtainReadPermission() was successfully called, and bytes were
successfully read.
*@param origByteCount is the originally requested number of bytes to get
permissions to read.
Modified:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ConnectionBin.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ConnectionBin.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ConnectionBin.java
(original)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ConnectionBin.java
Sun Jan 11 16:20:23 2015
@@ -124,8 +124,8 @@ public class ConnectionBin
* subsequent call to noteConnectionCreation() will be needed to confirm the
reservation, or clearReservation() to
* release the reservation.
*/
- public synchronized int waitConnectionAvailable(AtomicInteger poolCount)
- throws InterruptedException
+ public synchronized int waitConnectionAvailable(AtomicInteger poolCount,
IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
{
// Reserved connections keep a slot available which can't be used by
anyone else.
// Connection bins are always sorted so that deadlocks can't occur.
@@ -151,7 +151,16 @@ public class ConnectionBin
return IConnectionThrottler.CONNECTION_FROM_CREATION;
}
// Wait for a connection to free up. Note that it is up to the caller
to free stuff up.
- wait();
+ if (breakCheck == null)
+ {
+ wait();
+ }
+ else
+ {
+ long amt = breakCheck.abortCheck();
+ wait(amt);
+ }
+ // Back around
}
}
Modified:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/FetchBin.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/FetchBin.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/FetchBin.java
(original)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/FetchBin.java
Sun Jan 11 16:20:23 2015
@@ -100,8 +100,8 @@ public class FetchBin
* has permission to do the fetch, and can update the last fetch time.
*@return false if the fetch bin is being shut down.
*/
- public synchronized boolean reserveFetchRequest()
- throws InterruptedException
+ public synchronized boolean reserveFetchRequest(IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
{
// First wait for the ability to even get the next fetch from this bin
while (true)
@@ -113,7 +113,15 @@ public class FetchBin
reserveNextFetch = true;
return true;
}
- wait();
+ if (breakCheck == null)
+ {
+ wait();
+ }
+ else
+ {
+ long amt = breakCheck.abortCheck();
+ wait(amt);
+ }
}
}
@@ -131,9 +139,10 @@ public class FetchBin
* rights already, via reserveFetchRequest().
*@return false if the wait did not complete because the bin was shut down.
*/
- public synchronized boolean waitNextFetch()
- throws InterruptedException
+ public synchronized boolean waitNextFetch(IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
{
+ // MHL
if (!reserveNextFetch)
throw new IllegalStateException("No fetch request reserved!");
@@ -145,7 +154,16 @@ public class FetchBin
if (localMinimum == Long.MAX_VALUE)
{
// wait forever - but eventually someone will set a smaller interval
and wake us up.
- wait();
+ if (breakCheck == null)
+ {
+ wait();
+ }
+ else
+ {
+ long amt = breakCheck.abortCheck();
+ wait(amt);
+ }
+ // Back around
}
else
{
@@ -161,11 +179,22 @@ public class FetchBin
notifyAll();
return true;
}
- wait(waitAmt);
+ if (breakCheck == null)
+ {
+ wait(waitAmt);
+ }
+ else
+ {
+ long amt = breakCheck.abortCheck();
+ if (waitAmt < amt)
+ amt = waitAmt;
+ wait(amt);
+ }
+ // Back around
}
}
}
-
+
/** Poll this bin */
public synchronized void poll(IThreadContext threadContext)
throws ManifoldCFException
Modified:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ThrottleBin.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ThrottleBin.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ThrottleBin.java
(original)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/ThrottleBin.java
Sun Jan 11 16:20:23 2015
@@ -174,10 +174,9 @@ public class ThrottleBin
* read request takes place. Performs the necessary delay prior to reading
specified number of bytes from the server.
*@return false if the wait was interrupted due to the bin being shut down.
*/
- public boolean beginRead(int byteCount)
- throws InterruptedException
+ public boolean beginRead(int byteCount, IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
{
-
synchronized (this)
{
while (true)
@@ -186,7 +185,15 @@ public class ThrottleBin
return false;
if (estimateInProgress)
{
- wait();
+ if (breakCheck == null)
+ {
+ wait();
+ }
+ else
+ {
+ long amt = breakCheck.abortCheck();
+ wait(amt);
+ }
continue;
}
@@ -206,7 +213,15 @@ public class ThrottleBin
// If we haven't set a proper throttle yet, wait until we do.
if (localMinimum == Double.MAX_VALUE)
{
- wait();
+ if (breakCheck == null)
+ {
+ wait();
+ }
+ else
+ {
+ long amt = breakCheck.abortCheck();
+ wait(amt);
+ }
continue;
}
@@ -229,7 +244,17 @@ public class ThrottleBin
return true;
}
- this.wait(waitTime);
+ if (breakCheck == null)
+ {
+ this.wait(waitTime);
+ }
+ else
+ {
+ long amt = breakCheck.abortCheck();
+ if (waitTime < amt)
+ amt = waitTime;
+ wait(amt);
+ }
// Back around again...
}
}
Modified:
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/Throttler.java
URL:
http://svn.apache.org/viewvc/manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/Throttler.java?rev=1650911&r1=1650910&r2=1650911&view=diff
==============================================================================
---
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/Throttler.java
(original)
+++
manifoldcf/trunk/framework/connector-common/src/main/java/org/apache/manifoldcf/connectorcommon/throttler/Throttler.java
Sun Jan 11 16:20:23 2015
@@ -422,8 +422,8 @@ public class Throttler
* are available in the current pool, across all bins.
*@return the IConnectionThrottler codes for results.
*/
- public int waitConnectionAvailable(String[] binNames, AtomicInteger[]
poolCounts)
- throws InterruptedException
+ public int waitConnectionAvailable(String[] binNames, AtomicInteger[]
poolCounts, IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
{
// Each bin can signal something different. Bins that signal
// CONNECTION_FROM_NOWHERE are shutting down, but there's also
@@ -457,7 +457,7 @@ public class Throttler
int result;
try
{
- result = bin.waitConnectionAvailable(poolCounts[i]);
+ result = bin.waitConnectionAvailable(poolCounts[i],breakCheck);
}
catch (Throwable e)
{
@@ -472,6 +472,8 @@ public class Throttler
if (bin != null)
bin.undoReservation(currentRecommendation, poolCounts[i]);
}
+ if (e instanceof BreakException)
+ throw (BreakException)e;
if (e instanceof InterruptedException)
throw (InterruptedException)e;
if (e instanceof Error)
@@ -701,8 +703,8 @@ public class Throttler
*@param binNames are the names of the bins.
*@return false if being shut down
*/
- public boolean obtainFetchDocumentPermission(String[] binNames)
- throws InterruptedException
+ public boolean obtainFetchDocumentPermission(String[] binNames,
IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
{
// First, make sure all the bins exist, and reserve a slot in each
int i = 0;
@@ -715,9 +717,28 @@ public class Throttler
bin = fetchBins.get(binName);
}
// Reserve a slot
- if (bin == null || !bin.reserveFetchRequest())
+ try
{
- // Release previous reservations, and return null
+ if (bin == null || !bin.reserveFetchRequest(breakCheck))
+ {
+ // Release previous reservations, and return null
+ while (i > 0)
+ {
+ i--;
+ binName = binNames[i];
+ synchronized (fetchBins)
+ {
+ bin = fetchBins.get(binName);
+ }
+ if (bin != null)
+ bin.clearReservation();
+ }
+ return false;
+ }
+ }
+ catch (BreakException e)
+ {
+ // Release previous reservations, and rethrow
while (i > 0)
{
i--;
@@ -729,7 +750,7 @@ public class Throttler
if (bin != null)
bin.clearReservation();
}
- return false;
+ throw e;
}
i++;
}
@@ -747,7 +768,26 @@ public class Throttler
}
if (bin != null)
{
- if (!bin.waitNextFetch())
+ try
+ {
+ if (!bin.waitNextFetch(breakCheck))
+ {
+ // Undo the reservations we haven't processed yet
+ while (i < binNames.length)
+ {
+ binName = binNames[i];
+ synchronized (fetchBins)
+ {
+ bin = fetchBins.get(binName);
+ }
+ if (bin != null)
+ bin.clearReservation();
+ i++;
+ }
+ return false;
+ }
+ }
+ catch (BreakException e)
{
// Undo the reservations we haven't processed yet
while (i < binNames.length)
@@ -761,7 +801,7 @@ public class Throttler
bin.clearReservation();
i++;
}
- return false;
+ throw e;
}
}
i++;
@@ -793,8 +833,8 @@ public class Throttler
*@param byteCount is the number of bytes to get permissions to read.
*@return true if the wait took place as planned, or false if the system is
being shut down.
*/
- public boolean obtainReadPermission(String[] binNames, int byteCount)
- throws InterruptedException
+ public boolean obtainReadPermission(String[] binNames, int byteCount,
IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
{
int i = 0;
while (i < binNames.length)
@@ -805,7 +845,26 @@ public class Throttler
{
bin = throttleBins.get(binName);
}
- if (bin == null || !bin.beginRead(byteCount))
+ try
+ {
+ if (bin == null || !bin.beginRead(byteCount, breakCheck))
+ {
+ // End bins we've already done, and exit
+ while (i > 0)
+ {
+ i--;
+ binName = binNames[i];
+ synchronized (throttleBins)
+ {
+ bin = throttleBins.get(binName);
+ }
+ if (bin != null)
+ bin.endRead(byteCount,0);
+ }
+ return false;
+ }
+ }
+ catch (BreakException e)
{
// End bins we've already done, and exit
while (i > 0)
@@ -819,7 +878,7 @@ public class Throttler
if (bin != null)
bin.endRead(byteCount,0);
}
- return false;
+ throw e;
}
i++;
}
@@ -989,7 +1048,28 @@ public class Throttler
public int waitConnectionAvailable()
throws InterruptedException
{
- return parent.waitConnectionAvailable(binNames, poolCounts);
+ try
+ {
+ return waitConnectionAvailable(null);
+ }
+ catch (BreakException e)
+ {
+ throw new RuntimeException("Unexpected break exception:
"+e.getMessage(),e);
+ }
+ }
+
+ /** Get permission to grab a connection for use. If this object believes
there is a connection
+ * available in the pool, it will update its pool size variable and return
If not, this method
+ * evaluates whether a new connection should be created. If neither
condition is true, it
+ * waits until a connection is available.
+ *@return whether to take the connection from the pool, or create one, or
whether the
+ * throttler is being shut down.
+ */
+ @Override
+ public int waitConnectionAvailable(IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
+ {
+ return parent.waitConnectionAvailable(binNames, poolCounts, breakCheck);
}
/** For a new connection, obtain the fetch throttler to use for the
connection.
@@ -1088,7 +1168,26 @@ public class Throttler
public boolean obtainFetchDocumentPermission()
throws InterruptedException
{
- return parent.obtainFetchDocumentPermission(binNames);
+ try
+ {
+ return obtainFetchDocumentPermission(null);
+ }
+ catch (BreakException e)
+ {
+ throw new RuntimeException("Unexpected break exception:
"+e.getMessage(),e);
+ }
+ }
+
+ /** Get permission to fetch a document. This grants permission to start
+ * fetching a single document, within the connection that has already been
+ * granted permission that created this object.
+ *@return false if the throttler is being shut down.
+ */
+ @Override
+ public boolean obtainFetchDocumentPermission(IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
+ {
+ return parent.obtainFetchDocumentPermission(binNames,breakCheck);
}
/** Open a fetch stream. When done (or aborting), call
@@ -1128,9 +1227,30 @@ public class Throttler
public boolean obtainReadPermission(int byteCount)
throws InterruptedException
{
- return parent.obtainReadPermission(binNames, byteCount);
+ try
+ {
+ return obtainReadPermission(byteCount, null);
+ }
+ catch (BreakException e)
+ {
+ throw new RuntimeException("Unexpected break exception:
"+e.getMessage(),e);
+ }
}
+ /** Obtain permission to read a block of bytes. This method may wait
until it is OK to proceed.
+ * The throttle group, bin names, etc are already known
+ * to this specific interface object, so it is unnecessary to include them
here.
+ *@param byteCount is the number of bytes to get permissions to read.
+ *@param breakCheck is the break check object.
+ *@return true if the wait took place as planned, or false if the system is
being shut down.
+ */
+ @Override
+ public boolean obtainReadPermission(int byteCount, IBreakCheck breakCheck)
+ throws InterruptedException, BreakException
+ {
+ return parent.obtainReadPermission(binNames, byteCount, breakCheck);
+ }
+
/** Note the completion of the read of a block of bytes. Call this after
* obtainReadPermission() was successfully called, and bytes were
successfully read.
*@param origByteCount is the originally requested number of bytes to get
permissions to read.
