src/com/google/enterprise/adaptor/DocIdSender.java - plexi - Git at Google

 // Copyright 2011 Google Inc. All Rights Reserved.
 //
 // Licensed 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 com.google.enterprise.adaptor;

 import java.io.IOException;
 import java.util.*;
 import java.util.logging.*;

 /**
  * All logic for sending DocIds to the GSA from an adaptor.
  */
 class DocIdSender extends AbstractDocIdPusher
     implements AsyncDocIdSender.ItemPusher {
   private static final Logger log
       = Logger.getLogger(DocIdSender.class.getName());

   private final GsaFeedFileMaker fileMaker;
   private final GsaFeedFileSender fileSender;
   private final Journal journal;
   private final Config config;
   private final Adaptor adaptor;
   private final ExceptionHandler defaultErrorHandler
       = ExceptionHandlers.defaultHandler();

   public DocIdSender(GsaFeedFileMaker fileMaker, GsaFeedFileSender fileSender,
                      Journal journal, Config config, Adaptor adaptor) {
     this.fileMaker = fileMaker;
     this.fileSender = fileSender;
     this.journal = journal;
     this.config = config;
     this.adaptor = adaptor;
   }

   /**
    * Calls {@link Adaptor#getDocIds}. This method blocks until all DocIds are
    * sent or retrying failed.
    */
   public void pushFullDocIdsFromAdaptor(ExceptionHandler handler)
       throws InterruptedException {
     if (handler == null) {
       throw new NullPointerException();
     }
     log.info("Beginning getDocIds");
     journal.recordFullPushStarted();
     for (int ntries = 1;; ntries++) {
       boolean keepGoing = true;
       try {
         adaptor.getDocIds(this);
         break; // Success
       } catch (InterruptedException ex) {
         // Stop early.
         journal.recordFullPushInterrupted();
         log.info("Interrupted. Aborted getDocIds");
         throw ex;
       } catch (Exception ex) {
         log.log(Level.WARNING, "Exception during getDocIds", ex);
         keepGoing = handler.handleException(ex, ntries);
       }
       if (keepGoing) {
         log.log(Level.INFO, "Trying again... Number of attemps: {0}", ntries);
       } else {
         journal.recordFullPushFailed();
         log.warning("Gave up. Failed getDocIds");
         return; // Bail
       }
     }
     journal.recordFullPushSuccessful();
     log.info("Completed getDocIds");
   }

   /**
    * Calls {@link Adaptor#getModifiedDocIds}. This method blocks until all
    * DocIds are sent or retrying failed.
    */
   public void pushIncrementalDocIdsFromAdaptor(ExceptionHandler handler)
       throws InterruptedException {
     if (handler == null) {
       throw new NullPointerException();
     }
     log.info("Beginning getModifiedDocIds");
     journal.recordIncrementalPushStarted();
     for (int ntries = 1;; ntries++) {
       boolean keepGoing = true;
       try {
         ((PollingIncrementalAdaptor) adaptor).getModifiedDocIds(this);
         break; // Success
       } catch (InterruptedException ex) {
         // Stop early.
         journal.recordIncrementalPushInterrupted();
         log.info("Interrupted. Aborted getModifiedDocIds");
         throw ex;
       } catch (Exception ex) {
         log.log(Level.WARNING, "Exception during getModifiedDocIds", ex);
         keepGoing = handler.handleException(ex, ntries);
       }
       if (keepGoing) {
         log.log(Level.INFO, "Trying again... Number of attemps: {0}", ntries);
       } else {
         journal.recordIncrementalPushFailed();
         log.warning("Gave up. Failed getModifiedDocIds");
         return; // Bail
       }
     }
     journal.recordIncrementalPushSuccessful();
     log.info("Completed getModifiedDocIds");
   }

   /**
    * Makes and sends metadata-and-url feed files to GSA. Generally, you should
    * use {@link #pushDocIds()} instead of this method. However, if you want to
    * push just a few DocIds to the GSA manually, this is the method to use.
    * This method blocks until all DocIds are sent or retrying failed.
    */
   @Override
   public Record pushRecords(Iterable<Record> items, ExceptionHandler handler)
       throws InterruptedException {
     return pushItems(items.iterator(), handler);
   }

   @Override
   public DocId pushNamedResources(Map<DocId, Acl> resources,
                                   ExceptionHandler handler)
       throws InterruptedException {
     List<AclItem> acls = new ArrayList<AclItem>(resources.size());
     for (Map.Entry<DocId, Acl> me : resources.entrySet()) {
       acls.add(new AclItem(me.getKey(), me.getValue()));
     }
     AclItem acl = pushItems(acls.iterator(), handler);
     return acl == null ? null : acl.getDocId();
   }

   @Override
   public <T extends Item> T pushItems(Iterator<T> items,
       ExceptionHandler handler) throws InterruptedException {
     log.log(Level.INFO, "Pushing items");
     if (handler == null) {
       handler = defaultErrorHandler;
     }
     boolean firstBatch = true;
     final int max = config.getFeedMaxUrls();
     while (items.hasNext()) {
       List<T> batch = new ArrayList<T>();
       for (int j = 0; j < max; j++) {
         if (!items.hasNext()) {
           break;
         }
         batch.add(items.next());
       }
       log.log(Level.INFO, "Pushing group of {0} items", batch.size());
       T failedId;
       try {
         failedId = pushSizedBatchOfRecords(batch, handler);
       } catch (InterruptedException ex) {
         if (firstBatch) {
           throw ex;
         } else {
           // If this is not the first batch, then some items have already been
           // sent. Thus, return gracefully instead of throwing an exception so
           // that the caller can discover what was sent.
           Thread.currentThread().interrupt();
           return batch.get(0);
         }
       }
       if (failedId != null) {
         log.info("Failed to push all items. Failed on: " + failedId);
         return failedId;
       }
       firstBatch = false;
       journal.recordDocIdPush(batch);
     }
     log.info("Pushed items");
     return null;
   }

   public GroupPrincipal pushGroupDefinitions(
       Map<GroupPrincipal, ? extends Collection<Principal>> defs,
       boolean caseSensitive, ExceptionHandler handler)
       throws InterruptedException {
     if (defs.isEmpty()) {
       return null;
     }
     if (null == handler) {
       handler = defaultErrorHandler;
     }
     String groupsDefXml
         = fileMaker.makeGroupsDefinitionsXml(defs, caseSensitive);
     boolean keepGoing = true;
     boolean success = false;
     log.log(Level.INFO, "pushing groups");
     for (int ntries = 1; keepGoing; ntries++) {
       try {
         log.info("sending groups to GSA host name: " + config.getGsaHostname());
         fileSender.sendGroups(config.getFeedName(),
             groupsDefXml, config.isServerToUseCompression());
         keepGoing = false;  // Sent.
         success = true;
       } catch (IOException ex) {
         log.log(Level.WARNING, "failed to send groups", ex);
         keepGoing = handler.handleException(ex, ntries);
       }
       if (keepGoing) {
         log.log(Level.INFO, "trying again... number of attemps: {0}", ntries);
       }
     }
     GroupPrincipal last = null;
     if (success) {
       log.info("pushing groups succeeded");
     } else {
       log.log(Level.WARNING, "gave up pushing groups");
       last = defs.entrySet().iterator().next().getKey();  // checked on entry
     }
     log.info("finished pushing groups");
     return last;
   }


   private <T extends Item> T pushSizedBatchOfRecords(List<T> items,
                                          ExceptionHandler handler)
       throws InterruptedException {
     String feedSourceName = config.getFeedName();
     String xmlFeedFile = fileMaker.makeMetadataAndUrlXml(feedSourceName, items);
     boolean keepGoing = true;
     boolean success = false;
     log.log(Level.INFO, "Pushing batch of {0} items to GSA", items.size());
     for (int ntries = 1; keepGoing; ntries++) {
       try {
         log.info("Sending items to GSA host: " + config.getGsaHostname());
         fileSender.sendMetadataAndUrl(feedSourceName, xmlFeedFile,
                                       config.isServerToUseCompression());
         keepGoing = false;  // Sent.
         success = true;
       } catch (IOException ex) {
         log.log(Level.WARNING, "Failed to send items", ex);
         keepGoing = handler.handleException(ex, ntries);
       }
       if (keepGoing) {
         log.log(Level.INFO, "Trying again... Number of attemps: {0}", ntries);
       }
     }
     if (success) {
       log.info("Pushing batch succeeded");
     } else {
       log.log(Level.WARNING, "Gave up. First item in list: {0}", items.get(0));
     }
     log.info("Finished pushing batch of items");
     return success ? null : items.get(0);
   }

   /** Marker interface for an item that can exist in a feed. */
   interface Item {}

   /**
    * Represents the ACL tag sent in feeds.
    */
   static final class AclItem implements Item {
     private DocId id;
     private final String docIdFragment;
     private Acl acl;

     public AclItem(DocId id, Acl acl) {
       this(id, null, acl);
     }

     public AclItem(DocId id, String docIdFragment, Acl acl) {
       if (id == null || acl == null) {
         throw new NullPointerException("DocId and Acl must not be null");
       }
       this.id = id;
       this.docIdFragment = docIdFragment;
       this.acl = acl;
     }

     public DocId getDocId() {
       return id;
     }

     public String getDocIdFragment() {
       return docIdFragment;
     }

     public Acl getAcl() {
       return acl;
     }
   }
 }
	// Copyright 2011 Google Inc. All Rights Reserved.
	//
	// Licensed 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 com.google.enterprise.adaptor;

	import java.io.IOException;
	import java.util.*;
	import java.util.logging.*;

	/**
	* All logic for sending DocIds to the GSA from an adaptor.
	*/
	class DocIdSender extends AbstractDocIdPusher
	implements AsyncDocIdSender.ItemPusher {
	private static final Logger log
	= Logger.getLogger(DocIdSender.class.getName());

	private final GsaFeedFileMaker fileMaker;
	private final GsaFeedFileSender fileSender;
	private final Journal journal;
	private final Config config;
	private final Adaptor adaptor;
	private final ExceptionHandler defaultErrorHandler
	= ExceptionHandlers.defaultHandler();

	public DocIdSender(GsaFeedFileMaker fileMaker, GsaFeedFileSender fileSender,
	Journal journal, Config config, Adaptor adaptor) {
	this.fileMaker = fileMaker;
	this.fileSender = fileSender;
	this.journal = journal;
	this.config = config;
	this.adaptor = adaptor;
	}

	/**
	* Calls {@link Adaptor#getDocIds}. This method blocks until all DocIds are
	* sent or retrying failed.
	*/
	public void pushFullDocIdsFromAdaptor(ExceptionHandler handler)
	throws InterruptedException {
	if (handler == null) {
	throw new NullPointerException();
	}
	log.info("Beginning getDocIds");
	journal.recordFullPushStarted();
	for (int ntries = 1;; ntries++) {
	boolean keepGoing = true;
	try {
	adaptor.getDocIds(this);
	break; // Success
	} catch (InterruptedException ex) {
	// Stop early.
	journal.recordFullPushInterrupted();
	log.info("Interrupted. Aborted getDocIds");
	throw ex;
	} catch (Exception ex) {
	log.log(Level.WARNING, "Exception during getDocIds", ex);
	keepGoing = handler.handleException(ex, ntries);
	}
	if (keepGoing) {
	log.log(Level.INFO, "Trying again... Number of attemps: {0}", ntries);
	} else {
	journal.recordFullPushFailed();
	log.warning("Gave up. Failed getDocIds");
	return; // Bail
	}
	}
	journal.recordFullPushSuccessful();
	log.info("Completed getDocIds");
	}

	/**
	* Calls {@link Adaptor#getModifiedDocIds}. This method blocks until all
	* DocIds are sent or retrying failed.
	*/
	public void pushIncrementalDocIdsFromAdaptor(ExceptionHandler handler)
	throws InterruptedException {
	if (handler == null) {
	throw new NullPointerException();
	}
	log.info("Beginning getModifiedDocIds");
	journal.recordIncrementalPushStarted();
	for (int ntries = 1;; ntries++) {
	boolean keepGoing = true;
	try {
	((PollingIncrementalAdaptor) adaptor).getModifiedDocIds(this);
	break; // Success
	} catch (InterruptedException ex) {
	// Stop early.
	journal.recordIncrementalPushInterrupted();
	log.info("Interrupted. Aborted getModifiedDocIds");
	throw ex;
	} catch (Exception ex) {
	log.log(Level.WARNING, "Exception during getModifiedDocIds", ex);
	keepGoing = handler.handleException(ex, ntries);
	}
	if (keepGoing) {
	log.log(Level.INFO, "Trying again... Number of attemps: {0}", ntries);
	} else {
	journal.recordIncrementalPushFailed();
	log.warning("Gave up. Failed getModifiedDocIds");
	return; // Bail
	}
	}
	journal.recordIncrementalPushSuccessful();
	log.info("Completed getModifiedDocIds");
	}

	/**
	* Makes and sends metadata-and-url feed files to GSA. Generally, you should
	* use {@link #pushDocIds()} instead of this method. However, if you want to
	* push just a few DocIds to the GSA manually, this is the method to use.
	* This method blocks until all DocIds are sent or retrying failed.
	*/
	@Override
	public Record pushRecords(Iterable<Record> items, ExceptionHandler handler)
	throws InterruptedException {
	return pushItems(items.iterator(), handler);
	}

	@Override
	public DocId pushNamedResources(Map<DocId, Acl> resources,
	ExceptionHandler handler)
	throws InterruptedException {
	List<AclItem> acls = new ArrayList<AclItem>(resources.size());
	for (Map.Entry<DocId, Acl> me : resources.entrySet()) {
	acls.add(new AclItem(me.getKey(), me.getValue()));
	}
	AclItem acl = pushItems(acls.iterator(), handler);
	return acl == null ? null : acl.getDocId();
	}

	@Override
	public <T extends Item> T pushItems(Iterator<T> items,
	ExceptionHandler handler) throws InterruptedException {
	log.log(Level.INFO, "Pushing items");
	if (handler == null) {
	handler = defaultErrorHandler;
	}
	boolean firstBatch = true;
	final int max = config.getFeedMaxUrls();
	while (items.hasNext()) {
	List<T> batch = new ArrayList<T>();
	for (int j = 0; j < max; j++) {
	if (!items.hasNext()) {
	break;
	}
	batch.add(items.next());
	}
	log.log(Level.INFO, "Pushing group of {0} items", batch.size());
	T failedId;
	try {
	failedId = pushSizedBatchOfRecords(batch, handler);
	} catch (InterruptedException ex) {
	if (firstBatch) {
	throw ex;
	} else {
	// If this is not the first batch, then some items have already been
	// sent. Thus, return gracefully instead of throwing an exception so
	// that the caller can discover what was sent.
	Thread.currentThread().interrupt();
	return batch.get(0);
	}
	}
	if (failedId != null) {
	log.info("Failed to push all items. Failed on: " + failedId);
	return failedId;
	}
	firstBatch = false;
	journal.recordDocIdPush(batch);
	}
	log.info("Pushed items");
	return null;
	}

	public GroupPrincipal pushGroupDefinitions(
	Map<GroupPrincipal, ? extends Collection<Principal>> defs,
	boolean caseSensitive, ExceptionHandler handler)
	throws InterruptedException {
	if (defs.isEmpty()) {
	return null;
	}
	if (null == handler) {
	handler = defaultErrorHandler;
	}
	String groupsDefXml
	= fileMaker.makeGroupsDefinitionsXml(defs, caseSensitive);
	boolean keepGoing = true;
	boolean success = false;
	log.log(Level.INFO, "pushing groups");
	for (int ntries = 1; keepGoing; ntries++) {
	try {
	log.info("sending groups to GSA host name: " + config.getGsaHostname());
	fileSender.sendGroups(config.getFeedName(),
	groupsDefXml, config.isServerToUseCompression());
	keepGoing = false; // Sent.
	success = true;
	} catch (IOException ex) {
	log.log(Level.WARNING, "failed to send groups", ex);
	keepGoing = handler.handleException(ex, ntries);
	}
	if (keepGoing) {
	log.log(Level.INFO, "trying again... number of attemps: {0}", ntries);
	}
	}
	GroupPrincipal last = null;
	if (success) {
	log.info("pushing groups succeeded");
	} else {
	log.log(Level.WARNING, "gave up pushing groups");
	last = defs.entrySet().iterator().next().getKey(); // checked on entry
	}
	log.info("finished pushing groups");
	return last;
	}


	private <T extends Item> T pushSizedBatchOfRecords(List<T> items,
	ExceptionHandler handler)
	throws InterruptedException {
	String feedSourceName = config.getFeedName();
	String xmlFeedFile = fileMaker.makeMetadataAndUrlXml(feedSourceName, items);
	boolean keepGoing = true;
	boolean success = false;
	log.log(Level.INFO, "Pushing batch of {0} items to GSA", items.size());
	for (int ntries = 1; keepGoing; ntries++) {
	try {
	log.info("Sending items to GSA host: " + config.getGsaHostname());
	fileSender.sendMetadataAndUrl(feedSourceName, xmlFeedFile,
	config.isServerToUseCompression());
	keepGoing = false; // Sent.
	success = true;
	} catch (IOException ex) {
	log.log(Level.WARNING, "Failed to send items", ex);
	keepGoing = handler.handleException(ex, ntries);
	}
	if (keepGoing) {
	log.log(Level.INFO, "Trying again... Number of attemps: {0}", ntries);
	}
	}
	if (success) {
	log.info("Pushing batch succeeded");
	} else {
	log.log(Level.WARNING, "Gave up. First item in list: {0}", items.get(0));
	}
	log.info("Finished pushing batch of items");
	return success ? null : items.get(0);
	}

	/** Marker interface for an item that can exist in a feed. */
	interface Item {}

	/**
	* Represents the ACL tag sent in feeds.
	*/
	static final class AclItem implements Item {
	private DocId id;
	private final String docIdFragment;
	private Acl acl;

	public AclItem(DocId id, Acl acl) {
	this(id, null, acl);
	}

	public AclItem(DocId id, String docIdFragment, Acl acl) {
	if (id == null \|\| acl == null) {
	throw new NullPointerException("DocId and Acl must not be null");
	}
	this.id = id;
	this.docIdFragment = docIdFragment;
	this.acl = acl;
	}

	public DocId getDocId() {
	return id;
	}

	public String getDocIdFragment() {
	return docIdFragment;
	}

	public Acl getAcl() {
	return acl;
	}
	}
	}