/*
 * Copyright 2015, TeamDev Ltd. All rights reserved.
 *
 * Redistribution and use in source and/or binary forms, with or without
 * modification, must retain the above copyright notice and the following
 * disclaimer.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

package org.spine3.server.storage;

import com.google.common.collect.Lists;
import com.google.protobuf.Any;
import com.google.protobuf.TextFormat;
import org.spine3.TypeName;
import org.spine3.base.EventContext;
import org.spine3.base.EventId;
import org.spine3.base.EventRecord;
import org.spine3.server.aggregate.Snapshot;
import org.spine3.util.Events;
import org.spine3.util.Identifiers;

import java.util.Deque;
import java.util.Iterator;

/**
 * An event-sourced storage of aggregate root events and snapshots.
 *
 * @param <I> the type of IDs of aggregates managed by this storage
 * @author Alexander Yevsyukov
 */
public abstract class AggregateStorage<I> {

    public AggregateEvents load(I aggregateId) {
        Deque<EventRecord> history = Lists.newLinkedList();
        Snapshot snapshot = null;

        final Iterator<AggregateStorageRecord> historyBackward = historyBackward(aggregateId);
        while (historyBackward.hasNext()
                && snapshot == null) {
            AggregateStorageRecord record = historyBackward.next();
            switch (record.getKindCase()) {
                case EVENT_RECORD:
                    history.addFirst(record.getEventRecord());
                    break;
                case SNAPSHOT:
                    snapshot = record.getSnapshot();
                    break;
                case KIND_NOT_SET:
                    throw new IllegalStateException("Event record or snapshot missing in " + TextFormat.shortDebugString(record));
            }
        }

        AggregateEvents.Builder builder = AggregateEvents.newBuilder();
        if (snapshot != null) {
            builder.setSnapshot(snapshot);
        }
        builder.addAllEventRecord(history);

        return builder.build();
    }

    private static final String SNAPSHOT_TYPE_NAME = Snapshot.getDescriptor().getName();

    public void store(I aggregateId, Snapshot snapshot) {
        AggregateStorageRecord.Builder builder = AggregateStorageRecord.newBuilder()
                .setTimestamp(snapshot.getTimestamp())
                .setAggregateId(Identifiers.idToString(aggregateId))
                .setEventType(SNAPSHOT_TYPE_NAME)
                .setEventId("") // No event ID for snapshots
                .setVersion(snapshot.getVersion())
                .setSnapshot(snapshot);
        write(builder.build());
    }

    public void store(EventRecord record) {
        final EventContext context = record.getContext();
        final Any event = record.getEvent();
        final String aggregateId = Identifiers.idToString(context.getAggregateId());
        final EventId eventId = context.getEventId();
        final String eventIdStr = Identifiers.idToString(eventId);
        final String typeName = TypeName.ofEnclosed(event).nameOnly();
        AggregateStorageRecord.Builder builder = AggregateStorageRecord.newBuilder()
                .setTimestamp(Events.getTimestamp(eventId))
                .setAggregateId(aggregateId)
                .setEventType(typeName)
                .setEventId(eventIdStr)
                .setVersion(context.getVersion())
                .setEventRecord(record);
        write(builder.build());
    }

    // Storage implementation API.

    /**
     * Writes the passed record into the storage.
     *
     * @param record the record to write
     * @throws java.lang.NullPointerException if record or its aggregateId is null
     */
    protected abstract void write(AggregateStorageRecord record);

    /**
     * Creates iterator of aggregate event history with the reverse traversal.
     * Records are sorted by timestamp descending (from newer to older).
     *
     * @param id aggregate ID
     * @return new iterator instance, the iterator is empty if there's no history for the aggregate with passed ID
     */
    protected abstract Iterator<AggregateStorageRecord> historyBackward(I id);

    /**
     * Releases storage resources (closes I/O streams etc) if needed.
     */
    protected abstract void releaseResources();
}

@@ -1068,99 +1068,20 @@
 gle.
-common.collect.Lists;%0Aimport com.google.protobuf.Any;%0Aimport com.google.protobuf.TextFormat
+protobuf.Any
 ;%0Aim
@@ -1376,16 +1376,141 @@
 rator;%0A%0A
+import static com.google.common.collect.Lists.newLinkedList;%0Aimport static com.google.protobuf.TextFormat.shortDebugString;%0A%0A
 /**%0A * A
@@ -1769,24 +1769,25 @@
 regateId) %7B%0A
+%0A
         Dequ
@@ -1811,22 +1811,16 @@
 story = 
-Lists.
 newLinke
@@ -1954,24 +1954,25 @@
 gregateId);%0A
+%0A
         whil
@@ -2031,32 +2031,33 @@
 shot == null) %7B%0A
+%0A
             Aggr
@@ -2104,24 +2104,25 @@
 ard.next();%0A
+%0A
             
@@ -2517,23 +2517,50 @@
  in 
-%22 + TextFormat.
+record: %5C%22%22 +%0A                            
 shor
@@ -2579,16 +2579,23 @@
 (record)
+ + '%5C%22'
 );%0A     
@@ -2999,24 +2999,25 @@
 snapshot) %7B%0A
+%0A
         Aggr
@@ -3407,32 +3407,33 @@
 shot(snapshot);%0A
+%0A
         write(bu
@@ -3495,24 +3495,25 @@
 d record) %7B%0A
+%0A
         fina
@@ -3876,24 +3876,25 @@
 nameOnly();%0A
+%0A
         Aggr
@@ -4193,32 +4193,32 @@
 t.getVersion())%0A
-
                 
@@ -4238,24 +4238,25 @@
 rd(record);%0A
+%0A
         writ

Change exception message.