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);
		}
	}
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)