core/org.eclipse.smila.taskworker.test/code/src/org/eclipse/smila/taskworker/test/TestOutputs.java

/**********************************************************************************************************************
 * Copyright (c) 2008, 2011 Attensity Europe 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: Juergen Schumacher, Andreas Weber, Drazen Cindric, Andreas Schank (all Attensity Europe GmbH) - initial
 * implementation
 **********************************************************************************************************************/
package org.eclipse.smila.taskworker.test;

import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Map;

import org.eclipse.smila.datamodel.DataFactory;
import org.eclipse.smila.datamodel.Record;
import org.eclipse.smila.objectstore.ObjectStoreException;
import org.eclipse.smila.objectstore.ObjectStoreService;
import org.eclipse.smila.taskmanager.BulkInfo;
import org.eclipse.smila.taskworker.output.AppendableOutput;
import org.eclipse.smila.taskworker.output.Output;
import org.eclipse.smila.taskworker.output.Outputs;
import org.eclipse.smila.taskworker.output.RecordOutput;
import org.eclipse.smila.taskworker.output.StreamOutput;
import org.eclipse.smila.utils.collections.MultiValueMap;

/**
 * unit tests for the output wrappers.
 */
public class TestOutputs extends InputOutputTestbase {

  protected static final String STORE_NAME = "outputs";

  private Outputs _outputs;

  /** does nothing, just to test instantion. */
  public static class CustomOutput extends Output {

    /** input objects must provide this constructor. */
    public CustomOutput(final BulkInfo dataObject, final ObjectStoreService objectStore) {
      super(dataObject, objectStore);
    }

    @Override
    public void commit() throws ObjectStoreException, IOException {
      // nothing to do
    }

    @Override
    public void abort() throws IOException {
      // nothing to do.
    }

  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void setUp() throws Exception {
    super.setUp();
    _service.removeStore(STORE_NAME);
    _service.createStore(STORE_NAME, null);
    _outputs = new Outputs(prepareDataObjects(), _service);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void tearDown() throws Exception {
    super.tearDown();
  }

  /**
   * Writes one string to one slot and one object.
   * 
   * @throws Exception
   */
  public void testStreamOutputSingle() throws Exception {
    final StreamOutput streamOutput = _outputs.getAsStreamOutput("streamSingle");
    final String testString = "ThisIsAStreamOutputTest";
    streamOutput.getStream().write(testString.getBytes());
    streamOutput.commit();
    assertEquals(testString.length(), streamOutput.getBytesWritten());
    final byte[] result = _service.getObject(STORE_NAME, "streamSingleOutput");
    assertEquals(testString, new String(result));
  }

  /**
   * Writes several strings to one slot and several objects.
   * 
   * @throws Exception
   */
  public void testStreamOutputMulti() throws Exception {
    StreamOutput streamOutput = _outputs.getAsStreamOutput("streamMulti");
    String testString1 = "ThisIsAStreamOutputTest1 ";
    String testString2 = "ThisIsAStreamOutputTest2 ";
    String testString3 = "ThisIsAStreamOutputTest3 ";
    streamOutput.getStream().write(testString1.getBytes());
    streamOutput.getStream().write(testString2.getBytes());
    streamOutput.getStream().write(testString3.getBytes());
    streamOutput.commit();
    assertEquals(testString1.length() * 3, streamOutput.getBytesWritten());
    byte[] result = _service.getObject(STORE_NAME, "streamMultiOutput1");
    assertEquals(testString1 + testString2 + testString3, new String(result));
    streamOutput = _outputs.getAsStreamOutput("streamMulti", 1);
    testString1 = "ThisIsANewStreamOutputTest1 ";
    testString2 = "ThisIsANewStreamOutputTest2 ";
    testString3 = "ThisIsANewStreamOutputTest3 ";
    streamOutput.getStream().write(testString1.getBytes());
    streamOutput.getStream().write(testString2.getBytes());
    streamOutput.getStream().write(testString3.getBytes());
    streamOutput.commit();
    assertEquals(testString1.length() * 3, streamOutput.getBytesWritten());
    result = _service.getObject(STORE_NAME, "streamMultiOutput2");
    assertEquals(testString1 + testString2 + testString3, new String(result));
  }

  /**
   * Writes one record to one slot and one object.
   * 
   * @throws Exception
   */
  public void testRecordOutputSingle() throws Exception {
    final String objectName = "recordSingleOutput";
    final RecordOutput recordOutput = _outputs.getAsRecordOutput("recordSingle");
    final String title = "This is a title test";
    final Record r = DataFactory.DEFAULT.createRecord(title);
    r.getMetadata().put("title", title);
    recordOutput.writeRecord(r);
    recordOutput.commit();
    assertEquals(1, recordOutput.getRecordCount());
    assertTrue(recordOutput.getBytesWritten() > 0);
    final byte[] result = _service.getObject(STORE_NAME, objectName);
    final Record readRecord = IPCUTILS.binaryObject2record(result);
    assertEquals(r.getMetadata(), readRecord.getMetadata());
  }

  /**
   * Writes several records to one slot and several objects.
   * 
   * @throws Exception
   */
  public void testRecordOutputMulti() throws Exception {
    String objectName = "recordMultiOutput1";
    RecordOutput recordOutput = _outputs.getAsRecordOutput("recordMulti");
    String title1 = "This is a title test 1";
    Record r1 = DataFactory.DEFAULT.createRecord(title1);
    r1.getMetadata().put("title", title1);
    recordOutput.writeRecord(r1);
    String title2 = "This is a title test 2";
    Record r2 = DataFactory.DEFAULT.createRecord(title2);
    r2.getMetadata().put("title", title2);
    recordOutput.writeRecord(r2);
    recordOutput.commit();
    InputStream resultStream = _service.readObject(STORE_NAME, objectName);
    Record readRecord = IPCUTILS.binaryStream2record(resultStream);
    assertEquals(r1.getMetadata(), readRecord.getMetadata());
    readRecord = IPCUTILS.binaryStream2record(resultStream);
    assertEquals(r2.getMetadata(), readRecord.getMetadata());
    resultStream.close();
    assertEquals(2, recordOutput.getRecordCount());
    assertTrue(recordOutput.getBytesWritten() > 0);
    objectName = "recordMultiOutput2";
    recordOutput = _outputs.getAsRecordOutput("recordMulti", 1);
    title1 = "This is a new title test 1";
    r1 = DataFactory.DEFAULT.createRecord(title1);
    r1.getMetadata().put("title", title1);
    recordOutput.writeRecord(r1);
    title2 = "This is a new title test 2";
    r2 = DataFactory.DEFAULT.createRecord(title2);
    r2.getMetadata().put("title", title2);
    recordOutput.writeRecord(r2);
    recordOutput.commit();
    assertEquals(2, recordOutput.getRecordCount());
    assertTrue(recordOutput.getBytesWritten() > 0);
    resultStream = _service.readObject(STORE_NAME, objectName);
    readRecord = IPCUTILS.binaryStream2record(resultStream);
    assertEquals(r1.getMetadata(), readRecord.getMetadata());
    readRecord = IPCUTILS.binaryStream2record(resultStream);
    assertEquals(r2.getMetadata(), readRecord.getMetadata());
    resultStream.close();
  }

  /** test instantiation of a custom output class. */
  public void testCustomOutput() throws Exception {
    final CustomOutput custom = _outputs.getAsOutput("streamSingle", CustomOutput.class);
    assertNotNull(custom);
    assertEquals(STORE_NAME, custom.getStoreName());
    assertEquals("streamSingleOutput", custom.getObjectName());
  }

  /**
   * Writes plain data to appendable bulk.
   * 
   * @throws Exception
   */
  public void testAppendableOutput() throws Exception {
    final String objectName = "appendableSingle.app";
    final AppendableOutput appendable = _outputs.getAsOutput("appendableSingle", AppendableOutput.class);
    appendable.append("value1".getBytes());
    appendable.append("value2".getBytes());
    assertEquals(12, appendable.getBytesWritten());
    assertEquals(0, appendable.getRecordCount());
    assertEquals("value1value2", new String(_service.getObject(STORE_NAME, objectName)));
    appendable.append("value3".getBytes());
    appendable.commit();
    assertEquals(18, appendable.getBytesWritten());
    assertEquals(0, appendable.getRecordCount());
    assertEquals("value1value2value3", new String(_service.getObject(STORE_NAME, objectName)));
  }

  /**
   * Writes several records to appendable bulk.
   * 
   * @throws Exception
   */
  public void testAppendableRecordOutput() throws Exception {
    final String objectName = "appendableSingle.app";
    final AppendableOutput recordOutput = _outputs.getAsOutput("appendableSingle", AppendableOutput.class);
    final String title1 = "This is a title test 1";
    final Record r1 = DataFactory.DEFAULT.createRecord(title1);
    r1.getMetadata().put("title", title1);
    recordOutput.appendRecord(r1);
    final String title2 = "This is a title test 2";
    final Record r2 = DataFactory.DEFAULT.createRecord(title2);
    r2.getMetadata().put("title", title2);
    recordOutput.appendRecord(r2);
    recordOutput.commit();
    assertEquals(2, recordOutput.getRecordCount());
    assertTrue(recordOutput.getBytesWritten() > 0);
    final InputStream resultStream = _service.readObject(STORE_NAME, objectName);
    Record readRecord = IPCUTILS.binaryStream2record(resultStream);
    assertEquals(r1.getMetadata(), readRecord.getMetadata());
    readRecord = IPCUTILS.binaryStream2record(resultStream);
    assertEquals(r2.getMetadata(), readRecord.getMetadata());
  }

  /** setup input data object map. */
  private Map<String, List<BulkInfo>> prepareDataObjects() {
    final MultiValueMap<String, BulkInfo> dataObjects = new MultiValueMap<String, BulkInfo>();
    dataObjects.add("streamSingle", new BulkInfo("x", "outputs", "streamSingleOutput"));
    dataObjects.add("streamMulti", new BulkInfo("x", "outputs", "streamMultiOutput1"));
    dataObjects.add("streamMulti", new BulkInfo("x", "outputs", "streamMultiOutput2"));
    dataObjects.add("kvoSingle", new BulkInfo("x", "outputs", "kvoSingleOutput.kvo"));
    dataObjects.add("kvoMulti", new BulkInfo("x", "outputs", "kvoMultiOutput1.kvo"));
    dataObjects.add("kvoMulti", new BulkInfo("x", "outputs", "kvoMultiOutput2.kvo"));
    dataObjects.add("recordSingle", new BulkInfo("x", "outputs", "recordSingleOutput"));
    dataObjects.add("recordMulti", new BulkInfo("x", "outputs", "recordMultiOutput1"));
    dataObjects.add("recordMulti", new BulkInfo("x", "outputs", "recordMultiOutput2"));
    dataObjects.add("appendableSingle", new BulkInfo("x", "outputs", "appendableSingle.app"));
    return dataObjects;
  }
}