[java-metadata-aggregator] branch maint-0.9 updated: MDA-76 multi-output serialiser for offline use cases
Ian Young
ian at iay.org.uk
Fri Aug 26 12:13:05 EDT 2016
This is an automated email from the git hooks/post-receive script.
iay pushed a commit to branch maint-0.9
in repository java-metadata-aggregator.
View the commit online:
http://git.shibboleth.net/view/?p=java-metadata-aggregator.git;a=commit;h=ee5aa1bef86bbf307a68b31b566e9484aedc5e4b
The following commit(s) were added to refs/heads/maint-0.9 by this push:
new ee5aa1b MDA-76 multi-output serialiser for offline use cases
ee5aa1b is described below
commit ee5aa1bef86bbf307a68b31b566e9484aedc5e4b
Author: Ian Young <ian at iay.org.uk>
AuthorDate: Fri Aug 26 17:12:56 2016 +0100
MDA-76 multi-output serialiser for offline use cases
Add MultiOutputSerializationStage and basic in-memory unit test.
---
.../pipeline/MultiOutputSerializationStage.java | 177 +++++++++++++++++++++
.../MultiOutputSerializationStageTest.java | 102 ++++++++++++
2 files changed, 279 insertions(+)
diff --git a/aggregator-pipeline/src/main/java/net/shibboleth/metadata/pipeline/MultiOutputSerializationStage.java b/aggregator-pipeline/src/main/java/net/shibboleth/metadata/pipeline/MultiOutputSerializationStage.java
new file mode 100644
index 0000000..0c03f11
--- /dev/null
+++ b/aggregator-pipeline/src/main/java/net/shibboleth/metadata/pipeline/MultiOutputSerializationStage.java
@@ -0,0 +1,177 @@
+/*
+ * Licensed to the University Corporation for Advanced Internet Development,
+ * Inc. (UCAID) under one or more contributor license agreements. See the
+ * NOTICE file distributed with this work for additional information regarding
+ * copyright ownership. The UCAID licenses this file to You 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 net.shibboleth.metadata.pipeline;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Collection;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.ThreadSafe;
+
+import net.shibboleth.metadata.Item;
+import net.shibboleth.metadata.ItemSerializer;
+import net.shibboleth.utilities.java.support.annotation.constraint.NonnullElements;
+import net.shibboleth.utilities.java.support.component.ComponentInitializationException;
+import net.shibboleth.utilities.java.support.component.ComponentSupport;
+import net.shibboleth.utilities.java.support.logic.Constraint;
+
+/**
+ * A stage which writes the given item collection out to multiple destinations.
+ *
+ * The destination for each item is determined by a strategy function, and the
+ * serialisation is performed using a provided {@link ItemSerializer}.
+ *
+ * <p>
+ * This stage requires the following properties be set prior to initialization:
+ * <ul>
+ * <li><code>outputStrategy</code></li>
+ * <li><code>serializer</code></li>
+ * </ul>
+ *
+ * @param <T> type of items upon which this stage operates
+ */
+ at ThreadSafe
+public class MultiOutputSerializationStage<T> extends BaseStage<T> {
+
+ /**
+ * Interface implemented by destination objects created by a {@link Strategy}.
+ *
+ * The {@Destination} must be closed after the associated {@OutputStream}. This
+ * extra level of processing allows the {@Destination} to perform other tasks
+ * after the stream is closed without having to subclass or proxy the stream
+ * class.
+ */
+ public interface Destination extends Closeable {
+
+ /**
+ * Gets an {@link OutputStream} to which an item should be serialized.
+ *
+ * @return output stream to which to serialize the item
+ * @throws IOException if there is an error creating the output stream
+ */
+ @Nonnull OutputStream getOutputStream() throws IOException;
+ }
+
+ /**
+ * Interface to be implemented by strategy object determining where
+ * an item should be written.
+ *
+ * @param <T> type of items on which this strategy operates
+ */
+ public interface OutputStrategy<T> {
+
+ /**
+ * Gets a {@link Destination} to which an item should be serialized.
+ *
+ * @param item {@link Item} for which to generate a {@link Destination}
+ * @return {@link Destination} for the item
+ *
+ * @throws StageProcessingException if an output stream cannot be generated for
+ * the {@link Item}
+ */
+ @Nonnull Destination getDestination(@Nonnull final Item<T> item)
+ throws StageProcessingException;
+
+ }
+
+ /** Strategy used to determine where to serialize the item. */
+ private OutputStrategy<T> outputStrategy;
+
+ /** Serializer used to write the collection to the output stream. */
+ private ItemSerializer<T> serializer;
+
+ /**
+ * Gets the output strategy function determining where an item will be written.
+ *
+ * @return the output strategy function
+ */
+ @Nullable public OutputStrategy<T> getOutputStrategy() {
+ return outputStrategy;
+ }
+
+ /**
+ * Sets the output strategy function determining where an item will be written.
+ *
+ * @param strategy output strategy function determining where an item will be written
+ */
+ public synchronized void setOutputStrategy(@Nonnull final OutputStrategy<T> strategy) {
+ ComponentSupport.ifDestroyedThrowDestroyedComponentException(this);
+ ComponentSupport.ifInitializedThrowUnmodifiabledComponentException(this);
+
+ outputStrategy = Constraint.isNotNull(strategy, "Output strategy can not be null");
+ }
+
+ /**
+ * Gets the serializer used to write item to the output file.
+ *
+ * @return serializer used to write item to the output file
+ */
+ @Nullable public ItemSerializer<T> getSerializer() {
+ return serializer;
+ }
+
+ /**
+ * Sets the serializer used to write item to the output file.
+ *
+ * @param itemSerializer serializer used to write item to the output file
+ */
+ public synchronized void setSerializer(@Nonnull final ItemSerializer<T> itemSerializer) {
+ ComponentSupport.ifDestroyedThrowDestroyedComponentException(this);
+ ComponentSupport.ifInitializedThrowUnmodifiabledComponentException(this);
+
+ serializer = Constraint.isNotNull(itemSerializer, "Item serializer can not be null");
+ }
+
+ @Override
+ protected void doExecute(@Nonnull @NonnullElements Collection<Item<T>> itemCollection)
+ throws StageProcessingException {
+ for (final Item<T> item : itemCollection) {
+ try (final Destination destination = outputStrategy.getDestination(item);
+ final OutputStream stream = destination.getOutputStream()) {
+ serializer.serialize(item, stream);
+ } catch (IOException e) {
+ throw new StageProcessingException("Error writing to output location", e);
+ }
+ }
+ }
+
+ @Override
+ protected void doDestroy() {
+ outputStrategy = null;
+ serializer = null;
+
+ super.doDestroy();
+ }
+
+ @Override
+ protected void doInitialize() throws ComponentInitializationException {
+ super.doInitialize();
+
+ if (outputStrategy == null) {
+ throw new ComponentInitializationException("Output file can not be null");
+ }
+
+ if (serializer == null) {
+ throw new ComponentInitializationException("Item collection serializer can not be null");
+ }
+
+ }
+}
diff --git a/aggregator-pipeline/src/test/java/net/shibboleth/metadata/pipeline/MultiOutputSerializationStageTest.java b/aggregator-pipeline/src/test/java/net/shibboleth/metadata/pipeline/MultiOutputSerializationStageTest.java
new file mode 100644
index 0000000..98b80d3
--- /dev/null
+++ b/aggregator-pipeline/src/test/java/net/shibboleth/metadata/pipeline/MultiOutputSerializationStageTest.java
@@ -0,0 +1,102 @@
+
+package net.shibboleth.metadata.pipeline;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.testng.annotations.Test;
+
+import junit.framework.Assert;
+import net.shibboleth.metadata.Item;
+import net.shibboleth.metadata.ItemId;
+import net.shibboleth.metadata.ItemSerializer;
+import net.shibboleth.metadata.MockItem;
+import net.shibboleth.metadata.pipeline.MultiOutputSerializationStage.Destination;
+
+public class MultiOutputSerializationStageTest {
+
+ private static class StringMapOutputStrategy implements MultiOutputSerializationStage.OutputStrategy<String> {
+
+ public final Map<String, String> map = new HashMap<>();
+
+ private class StringDestination implements MultiOutputSerializationStage.Destination {
+
+ private final String id;
+ private final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+ public StringDestination(String i) {
+ id = i;
+ }
+
+ @Override
+ public void close() throws IOException {
+ map.put(id, baos.toString("UTF-8"));
+ }
+
+ @Override
+ public OutputStream getOutputStream() throws IOException {
+ return baos;
+ }
+
+ }
+
+ @Override
+ public Destination getDestination(Item<String> item) throws StageProcessingException {
+ // Locate the item's identifier.
+ final List<ItemId> ids = item.getItemMetadata().get(ItemId.class);
+ if (ids.isEmpty()) {
+ throw new StageProcessingException("item has no ItemId to base a file name on");
+ }
+ final ItemId id = ids.get(0);
+ return new StringDestination(id.getId());
+ }
+
+ }
+
+ @Test public void basicTest() throws Exception {
+ final Item<String> aaaItem = new MockItem("aaaContent");
+ aaaItem.getItemMetadata().put(new ItemId("aaa"));
+ final Item<String> bbbItem = new MockItem("bbbContent");
+ bbbItem.getItemMetadata().put(new ItemId("bbb"));
+ final Item<String> cccItem = new MockItem("cccContent");
+ cccItem.getItemMetadata().put(new ItemId("ccc"));
+ final List<Item<String>> items = new ArrayList<>();
+ items.add(aaaItem);
+ items.add(cccItem);
+ items.add(bbbItem);
+
+ final StringMapOutputStrategy strategy = new StringMapOutputStrategy();
+
+ final MultiOutputSerializationStage<String> stage = new MultiOutputSerializationStage<>();
+ stage.setId("test");
+ stage.setOutputStrategy(strategy);
+ stage.setSerializer(new ItemSerializer<String> () {
+
+ @Override
+ public void serialize(Item<String> item, OutputStream output) {
+ try {
+ output.write(item.unwrap().getBytes("UTF-8"));
+ } catch (IOException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+
+ });
+ stage.initialize();
+
+ stage.execute(items);
+
+ final Map<String, String> map = strategy.map;
+ Assert.assertEquals(3, map.size());
+ Assert.assertEquals("aaaContent", map.get("aaa"));
+ Assert.assertEquals("bbbContent", map.get("bbb"));
+ Assert.assertEquals("cccContent", map.get("ccc"));
+ }
+
+}
--
To stop receiving notification emails like this one, please contact
the administrator of this repository.
More information about the commits
mailing list