/* This file is part of VoltDB.
 * Copyright (C) 2020 VoltDB Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with VoltDB.  If not, see <http://www.gnu.org/licenses/>.
 */

package org.voltdb.e3;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;

import org.voltdb.messaging.VoltDbMessageFactory;

import com.google_voltpatches.common.collect.ImmutableList;
import com.google_voltpatches.common.collect.Range;

/**
 * A request for ranges of sequences from a specific stream and partition. The response to this is a
 * {@link GapFillResponse}.
 */
public class GapFillRequest extends GapFillMessage {
    private Collection<Range<Long>> m_ranges;

    public GapFillRequest() {
        super(VoltDbMessageFactory.E3_GAP_FILL_REQUEST);
    }

    /**
     * @param streamName
     * @param partitionId
     * @param start       inclusive start sequence number
     * @param end         inclusive end sequence number
     */
    GapFillRequest(String streamName, int partitionId, long start, long end) {
        this(streamName, partitionId, ImmutableList.of(Range.closed(start, end)));
    }

    /**
     * @param streamName
     * @param partitionId
     * @param ranges      {@link Collection} of {@link Range}s being requested. All ranges must be
     *                    {@link Range#closed(Comparable, Comparable)}
     */
    GapFillRequest(String streamName, int partitionId, Collection<Range<Long>> ranges) {
        super(VoltDbMessageFactory.E3_GAP_FILL_REQUEST, streamName, partitionId);
        m_ranges = ImmutableList.copyOf(ranges);
    }

    public Collection<Range<Long>> getRanges() {
        return m_ranges;
    }

    @Override
    public int getSerializedSize() {
        // super + rangeCount + ranges
        return super.getSerializedSize() + Integer.BYTES + m_ranges.size() * (Long.BYTES + Long.BYTES);
    }

    @Override
    protected void initFromBuffer(ByteBuffer buf) throws IOException {
        super.initFromBuffer(buf);
        int count = buf.getInt();
        ImmutableList.Builder<Range<Long>> builder = ImmutableList.builder();
        for (int i = 0; i < count; ++i) {
            builder.add(Range.closed(buf.getLong(), buf.getLong()));
        }
        m_ranges = builder.build();
    }

    @Override
    public void flattenToBuffer(ByteBuffer buf) throws IOException {
        super.flattenToBuffer(buf);
        buf.putInt(m_ranges.size());
        for (Range<Long> range : m_ranges) {
            buf.putLong(range.lowerEndpoint());
            buf.putLong(range.upperEndpoint());
        }
    }
}

@@ -838,16 +838,70 @@
 ction;%0A%0A
+import org.voltdb.export.ExportSequenceNumberTracker;%0A
 import o
@@ -942,16 +942,16 @@
 actory;%0A
-
 %0Aimport 
@@ -1411,32 +1411,48 @@
 param streamName
+  Name of stream
 %0A     * @param p
@@ -1453,32 +1453,77 @@
 aram partitionId
+ partition id ranges should be retrieved from
 %0A     * @param s
@@ -1830,16 +1830,32 @@
 reamName
+  Name of stream
 %0A     * 
@@ -1872,16 +1872,61 @@
 titionId
+ partition id ranges should be retrieved from
 %0A     * 
@@ -2000,99 +2000,8 @@
 sted
-. All ranges must be%0A     *                    %7B@link Range#closed(Comparable, Comparable)%7D
 %0A   
@@ -3139,76 +3139,112 @@
 ong(
-range.lowerEndpoint());%0A            buf.putLong(range.upperEndpoint(
+ExportSequenceNumberTracker.start(range));%0A            buf.putLong(ExportSequenceNumberTracker.end(range
 ));%0A

Use utility methods to request closed ranges