core/org.eclipse.smila.jdbc/code/src/org/eclipse/smila/jdbc/JdbcLoggingPipelet.java

/*******************************************************************************
 * Copyright (c) 2008, 2013 Empolis Information Management GmbH and brox IT Solutions GmbH. All rights reserved. This
 * program and the accompanying materials are made available under the terms of the Eclipse Public License v1.0 which
 * accompanies this distribution, and is available at http://www.eclipse.org/legal/epl-v10.html
 * 
 * Contributors: Andreas Weber (Empolis Information Management GmbH) - initial API and implementation
 *******************************************************************************/
package org.eclipse.smila.jdbc;

import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.eclipse.smila.blackboard.Blackboard;
import org.eclipse.smila.datamodel.AnyMap;
import org.eclipse.smila.datamodel.Value;
import org.eclipse.smila.processing.ProcessingException;
import org.eclipse.smila.processing.parameters.ParameterAccessor;
import org.eclipse.smila.processing.util.ProcessingConstants;
import org.eclipse.smila.processing.util.ResultCollector;
import org.eclipse.smila.utils.service.ServiceUtils;

/**
 * Pipelet that logs given statement (PreparedStatement) into a database via {@link JdbcWriterService} for each input
 * record.
 * 
 * The values for the PreparedStatement are selected from the input record by following the configured value paths.
 * 
 * If a value path references a single value, this is used. If it references a sequence of values, the first value is
 * used. In any other case, the value is set 'null'.
 */
public class JdbcLoggingPipelet extends AbstractJdbcPipelet {

  /** local logger. */
  private final Log _log = LogFactory.getLog(getClass());

  /** OSGI service used to log to database. */
  private JdbcWriterService _jdbcWriter;

  @Override
  public String[] process(final Blackboard blackboard, final String[] recordIds) throws ProcessingException {
    final JdbcWriterService jdbcWriterService = getJdbcWriterService();
    final ParameterAccessor paramAccessor = new ParameterAccessor(blackboard, _configuration);
    final ResultCollector resultCollector =
      new ResultCollector(paramAccessor, _log, ProcessingConstants.DROP_ON_ERROR_DEFAULT);
    if (recordIds != null) {
      for (final String id : recordIds) {
        try {
          paramAccessor.setCurrentRecord(id);
          final String dbUrl = getDbUrl(paramAccessor);
          final AnyMap dbProps = getDbProps(paramAccessor);
          final String stmt = getStatement(paramAccessor);
          final List<Value> stmtValues = getStatementValues(paramAccessor, blackboard.getMetadata(id));
          jdbcWriterService.write(dbUrl, dbProps, stmt, stmtValues);
          resultCollector.addResult(id);
        } catch (final Exception e) {
          resultCollector.addFailedResult(id, e);
        }
      }
    }
    return resultCollector.getResultIds();
  }

  /** @return a {@link JdbcWriterService} service. */
  private synchronized JdbcWriterService getJdbcWriterService() throws ProcessingException {
    if (_jdbcWriter == null) {
      try {
        _jdbcWriter = ServiceUtils.getService(JdbcWriterService.class);
      } catch (final Exception ex) {
        _log.warn("Error while waiting for JdbcWriterService service to come up.", ex);
      }
      if (_jdbcWriter == null) {
        throw new ProcessingException("No JdbcWriterService service available, giving up");
      }
    }
    return _jdbcWriter;
  }
}