stream/proto/src/main/java/org/apache/bookkeeper/stream/protocol/util/ProtoUtils.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF 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 org.apache.bookkeeper.stream.protocol.util;

import static com.google.common.base.Preconditions.checkArgument;

import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import java.util.List;
import org.apache.bookkeeper.common.util.Revisioned;
import org.apache.bookkeeper.stream.proto.NamespaceConfiguration;
import org.apache.bookkeeper.stream.proto.RangeProperties;
import org.apache.bookkeeper.stream.proto.StreamConfiguration;
import org.apache.bookkeeper.stream.proto.StreamMetadata.LifecycleState;
import org.apache.bookkeeper.stream.proto.StreamMetadata.ServingState;
import org.apache.bookkeeper.stream.proto.StreamName;
import org.apache.bookkeeper.stream.proto.StreamProperties;
import org.apache.bookkeeper.stream.proto.storage.CreateNamespaceRequest;
import org.apache.bookkeeper.stream.proto.storage.CreateStreamRequest;
import org.apache.bookkeeper.stream.proto.storage.DeleteNamespaceRequest;
import org.apache.bookkeeper.stream.proto.storage.DeleteStreamRequest;
import org.apache.bookkeeper.stream.proto.storage.GetActiveRangesRequest;
import org.apache.bookkeeper.stream.proto.storage.GetNamespaceRequest;
import org.apache.bookkeeper.stream.proto.storage.GetStorageContainerEndpointRequest;
import org.apache.bookkeeper.stream.proto.storage.GetStorageContainerEndpointResponse;
import org.apache.bookkeeper.stream.proto.storage.GetStreamRequest;
import org.apache.bookkeeper.stream.proto.storage.OneStorageContainerEndpointRequest;
import org.apache.bookkeeper.stream.proto.storage.OneStorageContainerEndpointResponse;
import org.apache.bookkeeper.stream.proto.storage.StatusCode;
import org.apache.bookkeeper.stream.proto.storage.StorageContainerEndpoint;
import org.apache.commons.lang3.tuple.Pair;

/**
 * Protocol related utils.
 */
public class ProtoUtils {

    /**
     * Check if two key ranges overlaps with each other.
     *
     * @param meta1 first key range
     * @param meta2 second key range
     * @return true if two key ranges overlaps
     */
    public static boolean keyRangeOverlaps(RangeProperties meta1, RangeProperties meta2) {
        return keyRangeOverlaps(
            meta1.getStartHashKey(), meta1.getEndHashKey(),
            meta2.getStartHashKey(), meta2.getEndHashKey());
    }

    public static boolean keyRangeOverlaps(Pair<Long, Long> range1,
                                           Pair<Long, Long> range2) {
        return keyRangeOverlaps(range1.getLeft(), range1.getRight(),
            range2.getLeft(), range2.getRight());
    }

    public static boolean keyRangeOverlaps(RangeProperties range1,
                                           Pair<Long, Long> range2) {
        return keyRangeOverlaps(range1.getStartHashKey(), range1.getEndHashKey(),
            range2.getLeft(), range2.getRight());
    }

    public static boolean keyRangeOverlaps(Pair<Long, Long> range1,
                                           RangeProperties range2) {
        return keyRangeOverlaps(range1.getLeft(), range1.getRight(),
            range2.getStartHashKey(), range2.getEndHashKey());
    }

    static boolean keyRangeOverlaps(long startKey1,
                                    long endKey1,
                                    long startKey2,
                                    long endKey2) {
        return endKey2 > startKey1 && startKey2 < endKey1;
    }

    /**
     * Validate namespace name.
     *
     * @return true if it is a valid namespace name. otherwise false.
     */
    public static boolean validateNamespaceName(String name) {
        return validateStreamName(name);
    }

    /**
     * Validate stream name.
     *
     * <p>follow the rules that dlog uses for validating stream name.
     *
     * @return true if it is a valid namespace name. otherwise false.
     */
    public static boolean validateStreamName(String name) {
        if (Strings.isNullOrEmpty(name)) {
            return false;
        }
        for (int i = 0; i < name.length(); i++) {
            char c = name.charAt(i);
            if (c == 0
                || c == ' '
                || c == '<'
                || c == '>'
                || c > '\u0000' && c < '\u001f'
                || c > '\u007f' && c < '\u009f'
                || c > '\ud800' && c < '\uf8ff'
                || c > '\ufff0' && c < '\uffff') {
                return false;
            }
        }
        return true;
    }

    public static List<RangeProperties> split(long streamId,
                                              int numInitialRanges,
                                              long nextRangeId,
                                              StorageContainerPlacementPolicy placementPolicy) {
        int numRanges = Math.max(2, numInitialRanges);
        if (numRanges % 2 != 0) { // round up to odd number
            numRanges = numRanges + 1;
        }
        long rangeSize = Long.MAX_VALUE / (numRanges / 2);
        long startKey = Long.MIN_VALUE;
        List<RangeProperties> ranges = Lists.newArrayListWithExpectedSize(numRanges);
        for (int idx = 0; idx < numRanges; ++idx) {
            long endKey = startKey + rangeSize;
            if (numRanges - 1 == idx) {
                endKey = Long.MAX_VALUE;
            }
            long rangeId = nextRangeId++;
            RangeProperties props = RangeProperties.newBuilder()
                .setStartHashKey(startKey)
                .setEndHashKey(endKey)
                .setStorageContainerId(placementPolicy.placeStreamRange(streamId, rangeId))
                .setRangeId(rangeId)
                .build();
            startKey = endKey;

            ranges.add(props);
        }
        return ranges;
    }

    /**
     * Check if the stream is created.
     *
     * @param state stream state
     * @return true if the stream is in created state. otherwise, false.
     */
    public static boolean isStreamCreated(LifecycleState state) {
        checkArgument(state != LifecycleState.UNRECOGNIZED);
        return LifecycleState.UNINIT != state
            && LifecycleState.CREATING != state;
    }

    /**
     * Check if the stream is writable.
     *
     * @param state stream state
     * @return true if the stream is writable. otherwise, false.
     */
    public static boolean isStreamWritable(ServingState state) {
        checkArgument(state != ServingState.UNRECOGNIZED);
        return ServingState.WRITABLE == state;
    }

    //
    // Location API
    //

    /**
     * Create a {@link GetStorageContainerEndpointRequest}.
     *
     * @param storageContainers list of storage containers
     * @return a get storage container endpoint request.
     */
    public static GetStorageContainerEndpointRequest createGetStorageContainerEndpointRequest(
        List<Revisioned<Long>> storageContainers) {
        GetStorageContainerEndpointRequest.Builder builder = GetStorageContainerEndpointRequest.newBuilder();
        for (Revisioned<Long> storageContainer : storageContainers) {
            builder.addRequests(
                OneStorageContainerEndpointRequest.newBuilder()
                    .setStorageContainer(storageContainer.getValue())
                    .setRevision(storageContainer.getRevision()));
        }
        return builder.build();
    }

    /**
     * Create a {@link GetStorageContainerEndpointResponse}.
     *
     * @param endpoints list of storage container endpoints
     * @return a get storage container endpoint response.
     */
    public static GetStorageContainerEndpointResponse createGetStorageContainerEndpointResponse(
        List<StorageContainerEndpoint> endpoints) {
        GetStorageContainerEndpointResponse.Builder builder = GetStorageContainerEndpointResponse.newBuilder();
        builder.setStatusCode(StatusCode.SUCCESS);
        for (StorageContainerEndpoint endpoint : endpoints) {
            builder.addResponses(
                OneStorageContainerEndpointResponse.newBuilder()
                    .setStatusCode(StatusCode.SUCCESS)
                    .setEndpoint(endpoint));
        }
        return builder.build();
    }


    //
    // Meta Range API
    //

    public static GetActiveRangesRequest createGetActiveRangesRequest(long streamId) {
        return GetActiveRangesRequest.newBuilder()
                .setStreamId(streamId)
                .build();
    }

    public static GetActiveRangesRequest createGetActiveRangesRequest(StreamProperties streamProps) {
        return GetActiveRangesRequest.newBuilder()
                .setStreamId(streamProps.getStreamId())
                .setStreamProps(streamProps)
                .build();
    }

    //
    // Namespace API
    //

    /**
     * Create a {@link CreateNamespaceRequest}.
     *
     * @param nsName namespace name
     * @param nsConf namespace conf
     * @return a create namespace request.
     */
    public static CreateNamespaceRequest createCreateNamespaceRequest(String nsName,
                                                                      NamespaceConfiguration nsConf) {
        return CreateNamespaceRequest.newBuilder()
            .setName(nsName)
            .setNsConf(nsConf)
            .build();
    }

    /**
     * Create a {@link DeleteNamespaceRequest}.
     *
     * @param colName namespace name
     * @return a delete namespace request.
     */
    public static DeleteNamespaceRequest createDeleteNamespaceRequest(String colName) {
        return DeleteNamespaceRequest.newBuilder()
            .setName(colName)
            .build();
    }

    /**
     * Create a {@link GetNamespaceRequest}.
     *
     * @param colName namespace name
     * @return a get namespace request.
     */
    public static GetNamespaceRequest createGetNamespaceRequest(String colName) {
        return GetNamespaceRequest.newBuilder()
            .setName(colName)
            .build();
    }

    //
    // Stream API
    //

    /**
     * Create a {@link CreateStreamRequest}.
     *
     * @param nsName     namespace name
     * @param streamName stream name
     * @param streamConf stream configuration
     * @return a create stream request.
     */
    public static CreateStreamRequest createCreateStreamRequest(String nsName,
                                                                String streamName,
                                                                StreamConfiguration streamConf) {
        return CreateStreamRequest.newBuilder()
            .setNsName(nsName)
            .setName(streamName)
            .setStreamConf(streamConf)
            .build();
    }

    /**
     * Create a {@link GetStreamRequest}.
     *
     * @param nsName     namespace name
     * @param streamName stream name
     * @return a create stream request.
     */
    public static GetStreamRequest createGetStreamRequest(String nsName,
                                                          String streamName) {
        return GetStreamRequest.newBuilder()
            .setStreamName(StreamName.newBuilder()
                .setNamespaceName(nsName)
                .setStreamName(streamName))
            .build();
    }

    /**
     * Create a {@link GetStreamRequest}.
     *
     * @param streamId stream id
     * @return a create stream request.
     */
    public static GetStreamRequest createGetStreamRequest(long streamId) {
        return GetStreamRequest.newBuilder()
            .setStreamId(streamId)
            .build();
    }

    /**
     * Create a {@link DeleteStreamRequest}.
     *
     * @param nsName     namespace name
     * @param streamName stream name
     * @return a create stream request.
     */
    public static DeleteStreamRequest createDeleteStreamRequest(String nsName,
                                                                String streamName) {
        return DeleteStreamRequest.newBuilder()
            .setName(streamName)
            .setNsName(nsName)
            .build();
    }

}