001/*
002 * (C) Copyright 2018 Nuxeo SA (http://nuxeo.com/) and others.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *     http://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 *
016 * Contributors:
017 *     bdelbosc
018 */
019package org.nuxeo.lib.stream.log.kafka;
020
021/**
022 * Handle a namespace for Kafka topic and group.
023 *
024 * @since 10.1
025 */
026public class KafkaNamespace {
027
028    protected final String prefix;
029
030    protected final int prefixLen;
031
032    public KafkaNamespace(String prefix) {
033        this.prefix = prefix;
034        this.prefixLen = prefix.length();
035    }
036
037    public String getTopicName(String logName) {
038        return prefix + logName;
039    }
040
041    public String getLogName(String topicName) {
042        if (!topicName.startsWith(prefix)) {
043            throw new IllegalArgumentException(String.format("topic %s with invalid prefix %s", topicName, prefix));
044        }
045        return topicName.substring(prefixLen);
046    }
047
048    public String getKafkaGroup(String group) {
049        return prefix + group;
050    }
051
052    public String getGroup(String kafkaGroup) {
053        if (!kafkaGroup.startsWith(prefix)) {
054            throw new IllegalArgumentException(String.format("group %s with invalid prefix %s", kafkaGroup, prefix));
055        }
056        return kafkaGroup.substring(prefixLen);
057    }
058
059    @Override
060    public String toString() {
061        return "KafkaNamespace{" + "prefix='" + prefix + '\'' + '}';
062    }
063}