extensions/search/src/main/java/uk/co/tfd/sm/search/es/ContentEventListener.java
public void handleEvent(Event event) {
		String topic = event.getTopic();
		Session session = (Session) event.getProperty(Session.class.getName());
		RepositorySession repositoryRession = null;
		Thread thisThread = Thread.currentThread();
		ClassLoader classloader = thisThread.getContextClassLoader();
		// ES might load classes so we had better set the context classloader.
		try {
			thisThread.setContextClassLoader(this.getClass().getClassLoader());
			try {
				repositoryRession = new RepositorySessionImpl(session,
						repository);
			} catch (ClientPoolException e1) {
				LOGGER.error(e1.getMessage(), e1);
				return;
			} catch (StorageClientException e1) {
				LOGGER.error(e1.getMessage(), e1);
				return;
			} catch (AccessDeniedException e1) {
				LOGGER.error(e1.getMessage(), e1);
				return;
			}

			LOGGER.debug("Got Event {} {} ", event, handlers);
			Collection<IndexingHandler> contentIndexHandler = handlers
					.get(topic);
			if (contentIndexHandler != null && contentIndexHandler.size() > 0) {
				BulkRequestBuilder bulk = client.prepareBulk();
				int added = 0;
				for (IndexingHandler indexingHandler : contentIndexHandler) {
					Collection<InputDocument> documents = indexingHandler
							.getDocuments(repositoryRession, event);
					for (InputDocument in : documents) {
						LOGGER.info("Indexing {} ", in);
						if (in.isDelete()) {
							bulk.add(client.prepareDelete(in.getIndexName(),
									in.getDocumentType(), in.getDocumentId()));
							added++;
						} else {
							try {
								IndexRequestBuilder r = client.prepareIndex(
										in.getIndexName(),
										in.getDocumentType(),
										in.getDocumentId());
								XContentBuilder d = XContentFactory
										.jsonBuilder();
								d = d.startObject();
								for (Entry<String, Object> e : in.getKeyData()) {
									d = d.field(e.getKey(), e.getValue());
								}
								r.setSource(d.endObject());
								bulk.add(r);
								added++;
							} catch (IOException e) {
								LOGGER.error(e.getMessage(), e);
							}
						}
					}
				}
				if ( added > 0 ) {
					BulkResponse resp = bulk.execute().actionGet();
					if (resp.hasFailures()) {
						for (BulkItemResponse br : Iterables.adaptTo(resp
								.iterator())) {
							if (br.failed()) {
								LOGGER.error("Failed {} {} ", br.getId(),
										br.getFailureMessage());
								// not going to retry at the moment, just log.
							}
						}
					}
				}

			}
		} finally {
			if (repositoryRession != null) {
				repositoryRession.logout();
			}
			thisThread.setContextClassLoader(classloader);
		}
	}
public void handleEvent(Event event) {
		String topic = event.getTopic();
		Session session = (Session) event.getProperty(Session.class.getName());
		RepositorySession repositoryRession = null;
		Thread thisThread = Thread.currentThread();
		ClassLoader classloader = thisThread.getContextClassLoader();
		// ES might load classes so we had better set the context classloader.
		try {
			thisThread.setContextClassLoader(this.getClass().getClassLoader());
			try {
				repositoryRession = new RepositorySessionImpl(session,
						repository);
			} catch (ClientPoolException e1) {
				LOGGER.error(e1.getMessage(), e1);
				return;
			} catch (StorageClientException e1) {
				LOGGER.error(e1.getMessage(), e1);
				return;
			} catch (AccessDeniedException e1) {
				LOGGER.error(e1.getMessage(), e1);
				return;
			}

			LOGGER.debug("Got Event {} {} ", event, handlers);
			Collection<IndexingHandler> contentIndexHandler = handlers
					.get(topic);
			if (contentIndexHandler != null && contentIndexHandler.size() > 0) {
				BulkRequestBuilder bulk = client.prepareBulk();
				int added = 0;
				for (IndexingHandler indexingHandler : contentIndexHandler) {
					Collection<InputDocument> documents = indexingHandler
							.getDocuments(repositoryRession, event);
					for (InputDocument in : documents) {
						LOGGER.info("Indexing {} ", in);
						if (in.isDelete()) {
							bulk.add(client.prepareDelete(in.getIndexName(),
									in.getDocumentType(), in.getDocumentId()));
							added++;
						} else {
							try {
								IndexRequestBuilder r = client.prepareIndex(
										in.getIndexName(),
										in.getDocumentType(),
										in.getDocumentId());
								XContentBuilder d = XContentFactory
										.jsonBuilder();
								d = d.startObject();
								for (Entry<String, Object> e : in.getKeyData()) {
									d = d.field(e.getKey(), e.getValue());
								}
								r.setSource(d.endObject());
								bulk.add(r);
								added++;
							} catch (IOException e) {
								LOGGER.error(e.getMessage(), e);
							}
						}
					}
				}
				if ( added > 0 ) {
					BulkResponse resp = bulk.execute().actionGet();
					if (resp.hasFailures()) {
						for (BulkItemResponse br : Iterables.adaptTo(resp
								.iterator())) {
							if (br.failed()) {
								LOGGER.error("Failed {} {} ", br.getId(),
										br.getFailureMessage());
								// not going to retry at the moment, just log.
							}
						}
					}
				}

			}
		} finally {
			if (repositoryRession != null) {
				repositoryRession.logout();
			}
			thisThread.setContextClassLoader(classloader);
		}
	}
public void handleEvent ( Event event ) { String topic = event . getTopic ( ) ; Session session = ( Session ) event . getProperty ( Session . class . getName ( ) ) ; RepositorySession repositoryRession = null ; Thread thisThread = Thread . currentThread ( ) ; ClassLoader classloader = thisThread . getContextClassLoader ( ) ; // ES might load classes so we had better set the context classloader. try { thisThread . setContextClassLoader ( this . getClass ( ) . getClassLoader ( ) ) ; try { repositoryRession = new RepositorySessionImpl ( session , repository ) ; } catch ( ClientPoolException e1 ) { LOGGER . error ( e1 . getMessage ( ) , e1 ) ; return ; } catch ( StorageClientException e1 ) { LOGGER . error ( e1 . getMessage ( ) , e1 ) ; return ; } catch ( AccessDeniedException e1 ) { LOGGER . error ( e1 . getMessage ( ) , e1 ) ; return ; } LOGGER . debug ( "Got Event {} {} " , event , handlers ) ; Collection < IndexingHandler > contentIndexHandler = handlers . get ( topic ) ; if ( contentIndexHandler != null && contentIndexHandler . size ( ) > 0 ) { BulkRequestBuilder bulk = client . prepareBulk ( ) ; int added = 0 ; for ( IndexingHandler indexingHandler : contentIndexHandler ) { Collection < InputDocument > documents = indexingHandler . getDocuments ( repositoryRession , event ) ; for ( InputDocument in : documents ) { LOGGER . info ( "Indexing {} " , in ) ; if ( in . isDelete ( ) ) { bulk . add ( client . prepareDelete ( in . getIndexName ( ) , in . getDocumentType ( ) , in . getDocumentId ( ) ) ) ; added ++ ; } else { try { IndexRequestBuilder r = client . prepareIndex ( in . getIndexName ( ) , in . getDocumentType ( ) , in . getDocumentId ( ) ) ; XContentBuilder d = XContentFactory . jsonBuilder ( ) ; d = d . startObject ( ) ; for ( Entry < String , Object > e : in . getKeyData ( ) ) { d = d . field ( e . getKey ( ) , e . getValue ( ) ) ; } r . setSource ( d . endObject ( ) ) ; bulk . add ( r ) ; added ++ ; } catch ( IOException e ) { LOGGER . error ( e . getMessage ( ) , e ) ; } } } } if ( added > 0 ) { BulkResponse resp = bulk . execute ( ) . actionGet ( ) ; if ( resp . hasFailures ( ) ) { for ( BulkItemResponse br : Iterables . adaptTo ( resp . iterator ( ) ) ) { if ( br . failed ( ) ) { LOGGER . error ( "Failed {} {} " , br . getId ( ) , br . getFailureMessage ( ) ) ; // not going to retry at the moment, just log. } } } } } } finally { if ( repositoryRession != null ) { repositoryRession . logout ( ) ; } thisThread . setContextClassLoader ( classloader ) ; } }
Handles an event from OSGi and indexes it. The indexing operation should
only index metadata and not bodies. Indexing of bodies is performed by a
seperate thread.

@see org.osgi.service.event.EventHandler#handleEvent(org.osgi.service.event.Event)
Handles an event from OSGi and indexes it . The indexing operation should only index metadata and not bodies . Indexing of bodies is performed by a seperate thread .