manifoldcf-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kwri...@apache.org
Subject svn commit: r1634188 - in /manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch: AmazonCloudSearchConnector.java DocumentChunkManager.java DocumentRecord.java
Date Sat, 25 Oct 2014 07:38:24 GMT
Author: kwright
Date: Sat Oct 25 07:38:23 2014
New Revision: 1634188

URL: http://svn.apache.org/r1634188
Log:
Fix CONNECTORS-1077 for Amazon Cloud Search connector.

Modified:
    manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/AmazonCloudSearchConnector.java
    manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentChunkManager.java
    manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentRecord.java

Modified: manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/AmazonCloudSearchConnector.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/AmazonCloudSearchConnector.java?rev=1634188&r1=1634187&r2=1634188&view=diff
==============================================================================
--- manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/AmazonCloudSearchConnector.java
(original)
+++ manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/AmazonCloudSearchConnector.java
Sat Oct 25 07:38:23 2014
@@ -53,6 +53,7 @@ import org.apache.manifoldcf.agents.inte
 import org.apache.manifoldcf.agents.interfaces.IOutputNotifyActivity;
 import org.apache.manifoldcf.agents.interfaces.IOutputRemoveActivity;
 import org.apache.manifoldcf.agents.interfaces.IOutputCheckActivity;
+import org.apache.manifoldcf.agents.interfaces.IOutputHistoryActivity;
 import org.apache.manifoldcf.agents.interfaces.RepositoryDocument;
 import org.apache.manifoldcf.agents.interfaces.ServiceInterruption;
 import org.apache.manifoldcf.agents.output.BaseOutputConnector;
@@ -70,6 +71,7 @@ import org.apache.manifoldcf.core.interf
 import org.apache.manifoldcf.core.interfaces.BinaryInput;
 import org.apache.manifoldcf.core.interfaces.TempFileInput;
 import org.apache.manifoldcf.core.interfaces.VersionContext;
+import org.apache.manifoldcf.core.common.DateParser;
 import org.apache.manifoldcf.agents.system.ManifoldCF;
 import org.apache.manifoldcf.agents.system.Logging;
 
@@ -113,6 +115,9 @@ public class AmazonCloudSearchConnector 
   /** cloudsearch field name for file body text. */
   private static final String FILE_BODY_TEXT_FIELDNAME = "f_bodytext";
   
+  /** Field name we use for document's URI. */
+  private static final String DOCUMENT_URI_FIELDNAME = "document_URI";
+  
   /** Constructor.
    */
   public AmazonCloudSearchConnector(){
@@ -305,6 +310,7 @@ public class AmazonCloudSearchConnector 
   private final static Set<String> acceptableMimeTypes = new HashSet<String>();
   static
   {
+    // We presume input can be decoded using UTF-8, so we can accept only UTF-8 and others
for which this also applies
     acceptableMimeTypes.add("text/plain;charset=utf-8");
     acceptableMimeTypes.add("text/plain;charset=ascii");
     acceptableMimeTypes.add("text/plain;charset=us-ascii");
@@ -321,6 +327,8 @@ public class AmazonCloudSearchConnector 
   public boolean checkMimeTypeIndexable(VersionContext outputDescription, String mimeType,
IOutputCheckActivity activities)
     throws ManifoldCFException, ServiceInterruption
   {
+    if (mimeType == null)
+      return false;
     return acceptableMimeTypes.contains(mimeType.toLowerCase(Locale.ROOT));
   }
 
@@ -374,7 +382,7 @@ public class AmazonCloudSearchConnector 
       {
         for (int i = 0; i < elements.length; i++)
         {
-          elements[i] = new JSONStringReader(((Date)fieldValues[i]).toString());
+          elements[i] = new JSONStringReader(DateParser.formatISO8601Date((Date)fieldValues[i]));
         }
       }
       else if (fieldValues instanceof String[])
@@ -390,12 +398,16 @@ public class AmazonCloudSearchConnector 
       fieldReader.addNameValuePair(new JSONNameValueReader(new JSONStringReader(fieldName),new
JSONArrayReader(elements)));
     }
     
+    // Add in the original URI
+    fieldReader.addNameValuePair(new JSONNameValueReader(new JSONStringReader(DOCUMENT_URI_FIELDNAME),
+      new JSONStringReader(documentURI)));
+
     // Add the primary content data in.
     fieldReader.addNameValuePair(new JSONNameValueReader(new JSONStringReader(FILE_BODY_TEXT_FIELDNAME),
       new JSONStringReader(new InputStreamReader(document.getBinaryStream(),Consts.UTF_8))));
     
-    documentChunkManager.recordDocument(uid, serverHost, serverPath, new ReaderInputStream(objectReader,
Consts.UTF_8));
-    conditionallyFlushDocuments();
+    documentChunkManager.recordDocument(uid, serverHost, serverPath, documentURI, INGEST_ACTIVITY,
new Long(document.getBinaryLength()), new ReaderInputStream(objectReader, Consts.UTF_8));
+    conditionallyFlushDocuments(activities);
     return DOCUMENTSTATUS_ACCEPTED;
   }
   
@@ -423,32 +435,32 @@ public class AmazonCloudSearchConnector 
 
     try
     {
-      documentChunkManager.recordDocument(uid, serverHost, serverPath, new ReaderInputStream(objectReader,
Consts.UTF_8));
+      documentChunkManager.recordDocument(uid, serverHost, serverPath, documentURI, REMOVE_ACTIVITY,
null, new ReaderInputStream(objectReader, Consts.UTF_8));
     }
     catch (IOException e)
     {
       handleIOException(e);
     }
-    conditionallyFlushDocuments();
+    conditionallyFlushDocuments(activities);
   }
   
   @Override
   public void noteJobComplete(IOutputNotifyActivity activities)
       throws ManifoldCFException, ServiceInterruption {
     getSession();
-    flushDocuments();
+    flushDocuments(activities);
   }
   
   protected static final int CHUNK_SIZE = 1000;
 
-  protected void conditionallyFlushDocuments()
+  protected void conditionallyFlushDocuments(IOutputHistoryActivity activities)
     throws ManifoldCFException, ServiceInterruption
   {
     if (documentChunkManager.equalOrMoreThan(serverHost, serverPath, CHUNK_SIZE))
-      flushDocuments();
+      flushDocuments(activities);
   }
   
-  protected void flushDocuments()
+  protected void flushDocuments(IOutputHistoryActivity activities)
     throws ManifoldCFException, ServiceInterruption
   {
     Logging.ingest.info("AmazonCloudSearch: Starting flush to Amazon");
@@ -476,15 +488,43 @@ public class AmazonCloudSearchConnector 
         String status = getStatusFromJsonResponse(responsbody);
         if("success".equals(status))
         {
+          // Activity-log the individual documents we sent
+          for (DocumentRecord dr : records)
+          {
+            activities.recordActivity(null,dr.getActivity(),dr.getDataSize(),dr.getUri(),"OK",null);
+          }
           Logging.ingest.info("AmazonCloudSearch: Successfully sent document chunk " + chunkNumber);
           //remove documents from table..
           documentChunkManager.deleteChunk(records);
         }
         else
         {
-          Logging.ingest.error("AmazonCloudSearch: Error sending document chunk "+ chunkNumber+":
"+ responsbody);
-          throw new ManifoldCFException("recieved error status from service after feeding
document. response body : " + responsbody);
+          // Activity-log the individual documents that failed
+          for (DocumentRecord dr : records)
+          {
+            activities.recordActivity(null,dr.getActivity(),dr.getDataSize(),dr.getUri(),"FAILED",responsbody);
+          }
+          Logging.ingest.error("AmazonCloudSearch: Error sending document chunk "+ chunkNumber+":
'"+ responsbody + "'");
+          throw new ManifoldCFException("Received error status from service after feeding
document.  Response body: '" + responsbody +"'");
+        }
+      }
+      catch (ManifoldCFException e)
+      {
+        if (e.getErrorCode() == ManifoldCFException.INTERRUPTED)
+          throw e;
+        for (DocumentRecord dr : records)
+        {
+          activities.recordActivity(null,dr.getActivity(),dr.getDataSize(),dr.getUri(),e.getClass().getSimpleName().toUpperCase(Locale.ROOT),e.getMessage());
+        }
+        throw e;
+      }
+      catch (ServiceInterruption e)
+      {
+        for (DocumentRecord dr : records)
+        {
+          activities.recordActivity(null,dr.getActivity(),dr.getDataSize(),dr.getUri(),e.getClass().getSimpleName().toUpperCase(Locale.ROOT),e.getMessage());
         }
+        throw e;
       }
       finally
       {

Modified: manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentChunkManager.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentChunkManager.java?rev=1634188&r1=1634187&r2=1634188&view=diff
==============================================================================
--- manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentChunkManager.java
(original)
+++ manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentChunkManager.java
Sat Oct 25 07:38:23 2014
@@ -42,6 +42,9 @@ public class DocumentChunkManager extend
 {
   // Database fields
   private final static String UID_FIELD = "uid";                        // This is the document
key, which is a dochash value
+  private final static String URI_FIELD = "documenturi";            // This is the document
URI in plain text
+  private final static String ACTIVITY_FIELD = "activity";          //  This is a flag, for
activity logging, describing whether this is an indexing operation or a delete
+  private final static String LENGTH_FIELD = "doclength";          // Binary length of original
document
   private final static String HOST_FIELD = "serverhost";            // The host and path
are there to make sure we don't collide between connections
   private final static String PATH_FIELD = "serverpath";
   private final static String SDF_DATA_FIELD = "sdfdata";
@@ -68,12 +71,24 @@ public class DocumentChunkManager extend
         map.put(UID_FIELD,new ColumnDescription("VARCHAR(40)",false,false,null,null,false));
         map.put(HOST_FIELD,new ColumnDescription("VARCHAR(255)",false,false,null,null,false));
         map.put(PATH_FIELD,new ColumnDescription("VARCHAR(255)",false,false,null,null,false));
+        map.put(URI_FIELD,new ColumnDescription("LONGTEXT",false,true,null,null,false));
+        map.put(ACTIVITY_FIELD,new ColumnDescription("VARCHAR(255)",false,true,null,null,false));
+        map.put(LENGTH_FIELD,new ColumnDescription("BIGINT",false,true,null,null,false));
         map.put(SDF_DATA_FIELD,new ColumnDescription("BLOB",false,true,null,null,false));
         performCreate(map,null);
       }
       else
       {
         // Upgrade code, if needed, goes here
+        if (existing.get(URI_FIELD) == null)
+        {
+          // Add the new columns
+          HashMap map = new HashMap();
+          map.put(URI_FIELD,new ColumnDescription("LONGTEXT",false,true,null,null,false));
+          map.put(ACTIVITY_FIELD,new ColumnDescription("VARCHAR(255)",false,true,null,null,false));
+          map.put(LENGTH_FIELD,new ColumnDescription("BIGINT",false,true,null,null,false));
+          performAlter(map,null,null,null);
+        }
       }
 
       // Handle indexes, if needed
@@ -116,7 +131,7 @@ public class DocumentChunkManager extend
    * @param sdfData document SDF data.
    * @throws ManifoldCFException
    */
-  public void recordDocument(String uid, String host, String path, InputStream sdfData) 
+  public void recordDocument(String uid, String host, String path, String uri, String activity,
Long length, InputStream sdfData) 
       throws ManifoldCFException, IOException
   {
     TempFileInput tfi = null;
@@ -154,6 +169,10 @@ public class DocumentChunkManager extend
             
             Map<String,Object> parameterMap = new HashMap<String,Object>();
             parameterMap.put(SDF_DATA_FIELD, tfi);
+            parameterMap.put(URI_FIELD, uri);
+            parameterMap.put(ACTIVITY_FIELD, activity);
+            if (length != null)
+              parameterMap.put(LENGTH_FIELD, length);
             
             //if record exists on table, update record.
             if(set.getRowCount() > 0)
@@ -250,6 +269,9 @@ public class DocumentChunkManager extend
       IResultRow row = set.getRow(i);
       rval[i] = new DocumentRecord(host,path,
         (String)row.getValue(UID_FIELD),
+        (String)row.getValue(URI_FIELD),
+        (String)row.getValue(ACTIVITY_FIELD),
+        (Long)row.getValue(LENGTH_FIELD),
         (BinaryInput)row.getValue(SDF_DATA_FIELD));
     }
     return rval;

Modified: manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentRecord.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentRecord.java?rev=1634188&r1=1634187&r2=1634188&view=diff
==============================================================================
--- manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentRecord.java
(original)
+++ manifoldcf/trunk/connectors/amazoncloudsearch/connector/src/main/java/org/apache/manifoldcf/agents/output/amazoncloudsearch/DocumentRecord.java
Sat Oct 25 07:38:23 2014
@@ -27,13 +27,19 @@ public class DocumentRecord {
   protected final String host;
   protected final String path;
   protected final String uid;
+  protected final String uri;
+  protected final String activity;
+  protected final Long dataSize;
   protected final BinaryInput data;
   
-  public DocumentRecord(String host, String path, String uid, BinaryInput data)
+  public DocumentRecord(String host, String path, String uid, String uri, String activity,
Long dataSize, BinaryInput data)
   {
     this.host = host;
     this.path = path;
     this.uid = uid;
+    this.uri = uri;
+    this.activity = activity;
+    this.dataSize = dataSize;
     this.data = data;
   }
 
@@ -52,6 +58,21 @@ public class DocumentRecord {
     return uid;
   }
   
+  public String getUri()
+  {
+    return uri;
+  }
+  
+  public String getActivity()
+  {
+    return activity;
+  }
+  
+  public Long getDataSize()
+  {
+    return dataSize;
+  }
+  
   public long getStreamLength()
     throws ManifoldCFException
   {



Mime
View raw message