You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@manifoldcf.apache.org by kw...@apache.org on 2010/09/17 13:33:15 UTC
svn commit: r998081 - in /incubator/lcf/trunk/modules:
connectors/solr/connector/src/main/java/org/apache/acf/agents/output/solr/
framework/agents/src/main/java/org/apache/acf/agents/interfaces/
framework/agents/src/main/java/org/apache/acf/agents/outp...
Author: kwright
Date: Fri Sep 17 11:33:15 2010
New Revision: 998081
URL: http://svn.apache.org/viewvc?rev=998081&view=rev
Log:
Fix for CONNECTORS-106. Add output notification activity object to infrastructure so that core services can be provided to output connectors on job completion notification.
Added:
incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputNotifyActivity.java (with props)
Modified:
incubator/lcf/trunk/modules/connectors/solr/connector/src/main/java/org/apache/acf/agents/output/solr/SolrConnector.java
incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputConnector.java
incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/output/BaseOutputConnector.java
incubator/lcf/trunk/modules/framework/pull-agent/src/main/java/org/apache/acf/crawler/system/JobNotificationThread.java
Modified: incubator/lcf/trunk/modules/connectors/solr/connector/src/main/java/org/apache/acf/agents/output/solr/SolrConnector.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/modules/connectors/solr/connector/src/main/java/org/apache/acf/agents/output/solr/SolrConnector.java?rev=998081&r1=998080&r2=998081&view=diff
==============================================================================
--- incubator/lcf/trunk/modules/connectors/solr/connector/src/main/java/org/apache/acf/agents/output/solr/SolrConnector.java (original)
+++ incubator/lcf/trunk/modules/connectors/solr/connector/src/main/java/org/apache/acf/agents/output/solr/SolrConnector.java Fri Sep 17 11:33:15 2010
@@ -370,8 +370,9 @@ public class SolrConnector extends org.a
/** Notify the connector of a completed job.
* This is meant to allow the connector to flush any internal data structures it has been keeping around, or to tell the output repository that this
* is a good time to synchronize things. It is called whenever a job is either completed or aborted.
+ *@param activities is the handle to an object that the implementer of an output connector may use to perform operations, such as logging processing activity.
*/
- public void noteJobComplete()
+ public void noteJobComplete(IOutputNotifyActivity activities)
throws ACFException, ServiceInterruption
{
// Establish a session
Modified: incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputConnector.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputConnector.java?rev=998081&r1=998080&r2=998081&view=diff
==============================================================================
--- incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputConnector.java (original)
+++ incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputConnector.java Fri Sep 17 11:33:15 2010
@@ -134,8 +134,9 @@ public interface IOutputConnector extend
/** Notify the connector of a completed job.
* This is meant to allow the connector to flush any internal data structures it has been keeping around, or to tell the output repository that this
* is a good time to synchronize things. It is called whenever a job is either completed or aborted.
+ *@param activities is the handle to an object that the implementer of an output connector may use to perform operations, such as logging processing activity.
*/
- public void noteJobComplete()
+ public void noteJobComplete(IOutputNotifyActivity activities)
throws ACFException, ServiceInterruption;
// UI support methods.
Added: incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputNotifyActivity.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputNotifyActivity.java?rev=998081&view=auto
==============================================================================
--- incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputNotifyActivity.java (added)
+++ incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputNotifyActivity.java Fri Sep 17 11:33:15 2010
@@ -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.acf.agents.interfaces;
+
+import org.apache.acf.core.interfaces.*;
+import org.apache.acf.agents.interfaces.*;
+
+/** This interface abstracts from the activities that an output connector can do
+when processing notification of a completed job.
+*/
+public interface IOutputNotifyActivity extends IOutputHistoryActivity
+{
+ public static final String _rcsid = "@(#)$Id$";
+
+
+}
Propchange: incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputNotifyActivity.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/interfaces/IOutputNotifyActivity.java
------------------------------------------------------------------------------
svn:keywords = Id
Modified: incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/output/BaseOutputConnector.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/output/BaseOutputConnector.java?rev=998081&r1=998080&r2=998081&view=diff
==============================================================================
--- incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/output/BaseOutputConnector.java (original)
+++ incubator/lcf/trunk/modules/framework/agents/src/main/java/org/apache/acf/agents/output/BaseOutputConnector.java Fri Sep 17 11:33:15 2010
@@ -70,8 +70,9 @@ public abstract class BaseOutputConnecto
/** Notify the connector of a completed job.
* This is meant to allow the connector to flush any internal data structures it has been keeping around, or to tell the output repository that this
* is a good time to synchronize things. It is called whenever a job is either completed or aborted.
+ *@param activities is the handle to an object that the implementer of an output connector may use to perform operations, such as logging processing activity.
*/
- public void noteJobComplete()
+ public void noteJobComplete(IOutputNotifyActivity activities)
throws ACFException, ServiceInterruption
{
// The base implementation does nothing here.
Modified: incubator/lcf/trunk/modules/framework/pull-agent/src/main/java/org/apache/acf/crawler/system/JobNotificationThread.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/modules/framework/pull-agent/src/main/java/org/apache/acf/crawler/system/JobNotificationThread.java?rev=998081&r1=998080&r2=998081&view=diff
==============================================================================
--- incubator/lcf/trunk/modules/framework/pull-agent/src/main/java/org/apache/acf/crawler/system/JobNotificationThread.java (original)
+++ incubator/lcf/trunk/modules/framework/pull-agent/src/main/java/org/apache/acf/crawler/system/JobNotificationThread.java Fri Sep 17 11:33:15 2010
@@ -50,6 +50,7 @@ public class JobNotificationThread exten
IThreadContext threadContext = ThreadContextFactory.make();
IJobManager jobManager = JobManagerFactory.make(threadContext);
IOutputConnectionManager connectionManager = OutputConnectionManagerFactory.make(threadContext);
+ IRepositoryConnectionManager repositoryConnectionManager = RepositoryConnectionManagerFactory.make(threadContext);
// Loop
while (true)
@@ -69,8 +70,10 @@ public class JobNotificationThread exten
if (job != null)
{
// Get the connection name
- String connectionName = job.getOutputConnectionName();
- connectionNames.put(connectionName,connectionName);
+ String repositoryConnectionName = job.getConnectionName();
+ String outputConnectionName = job.getOutputConnectionName();
+ OutputAndRepositoryConnection c = new OutputAndRepositoryConnection(outputConnectionName, repositoryConnectionName);
+ connectionNames.put(c,c);
}
}
@@ -80,9 +83,14 @@ public class JobNotificationThread exten
Iterator iter = connectionNames.keySet().iterator();
while (iter.hasNext())
{
- String connectionName = (String)iter.next();
+ OutputAndRepositoryConnection connections = (OutputAndRepositoryConnection)iter.next();
- IOutputConnection connection = connectionManager.load(connectionName);
+ String outputConnectionName = connections.getOutputConnectionName();
+ String repositoryConnectionName = connections.getRepositoryConnectionName();
+
+ OutputNotifyActivity activity = new OutputNotifyActivity(repositoryConnectionName,repositoryConnectionManager,outputConnectionName);
+
+ IOutputConnection connection = connectionManager.load(outputConnectionName);
if (connection != null)
{
// Grab an appropriate connection instance
@@ -94,8 +102,8 @@ public class JobNotificationThread exten
// Do the notification itself
try
{
- connector.noteJobComplete();
- notifiedConnections.put(connectionName,connectionName);
+ connector.noteJobComplete(activity);
+ notifiedConnections.put(connections,connections);
}
catch (ServiceInterruption e)
{
@@ -132,8 +140,11 @@ public class JobNotificationThread exten
if (job != null)
{
// Get the connection name
- String connectionName = job.getOutputConnectionName();
- if (notifiedConnections.get(connectionName) != null)
+ String outputConnectionName = job.getOutputConnectionName();
+ String repositoryConnectionName = job.getConnectionName();
+ OutputAndRepositoryConnection c = new OutputAndRepositoryConnection(outputConnectionName, repositoryConnectionName);
+
+ if (notifiedConnections.get(c) != null)
{
// When done, put the job into the Inactive state. Otherwise, the notification will be retried until it succeeds.
jobManager.inactivateJob(jobID);
@@ -200,4 +211,83 @@ public class JobNotificationThread exten
}
}
+ /** Output connection/repository connection pair object */
+ protected static class OutputAndRepositoryConnection
+ {
+ protected String outputConnectionName;
+ protected String repositoryConnectionName;
+
+ public OutputAndRepositoryConnection(String outputConnectionName, String repositoryConnectionName)
+ {
+ this.outputConnectionName = outputConnectionName;
+ this.repositoryConnectionName = repositoryConnectionName;
+ }
+
+ public String getOutputConnectionName()
+ {
+ return outputConnectionName;
+ }
+
+ public String getRepositoryConnectionName()
+ {
+ return repositoryConnectionName;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (!(o instanceof OutputAndRepositoryConnection))
+ return false;
+ OutputAndRepositoryConnection x = (OutputAndRepositoryConnection)o;
+ return this.outputConnectionName.equals(x.outputConnectionName) && this.repositoryConnectionName.equals(x.repositoryConnectionName);
+ }
+
+ public int hashCode()
+ {
+ return outputConnectionName.hashCode() + repositoryConnectionName.hashCode();
+ }
+ }
+
+ /** The ingest logger class */
+ protected static class OutputNotifyActivity implements IOutputNotifyActivity
+ {
+
+ // Connection name
+ protected String connectionName;
+ // Connection manager
+ protected IRepositoryConnectionManager connMgr;
+ // Output connection name
+ protected String outputConnectionName;
+
+ /** Constructor */
+ public OutputNotifyActivity(String connectionName, IRepositoryConnectionManager connMgr, String outputConnectionName)
+ {
+ this.connectionName = connectionName;
+ this.connMgr = connMgr;
+ this.outputConnectionName = outputConnectionName;
+ }
+
+ /** Record time-stamped information about the activity of the output connector.
+ *@param startTime is either null or the time since the start of epoch in milliseconds (Jan 1, 1970). Every
+ * activity has an associated time; the startTime field records when the activity began. A null value
+ * indicates that the start time and the finishing time are the same.
+ *@param activityType is a string which is fully interpretable only in the context of the connector involved, which is
+ * used to categorize what kind of activity is being recorded. For example, a web connector might record a
+ * "fetch document" activity. Cannot be null.
+ *@param dataSize is the number of bytes of data involved in the activity, or null if not applicable.
+ *@param entityURI is a (possibly long) string which identifies the object involved in the history record.
+ * The interpretation of this field will differ from connector to connector. May be null.
+ *@param resultCode contains a terse description of the result of the activity. The description is limited in
+ * size to 255 characters, and can be interpreted only in the context of the current connector. May be null.
+ *@param resultDescription is a (possibly long) human-readable string which adds detail, if required, to the result
+ * described in the resultCode field. This field is not meant to be queried on. May be null.
+ */
+ public void recordActivity(Long startTime, String activityType, Long dataSize,
+ String entityURI, String resultCode, String resultDescription)
+ throws ACFException
+ {
+ connMgr.recordHistory(connectionName,startTime,ACF.qualifyOutputActivityName(activityType,outputConnectionName),dataSize,entityURI,resultCode,
+ resultDescription,null);
+ }
+
+ }
}