001/*
002 * (C) Copyright 2017 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.tools.command;
020
021import java.time.Duration;
022
023import org.apache.commons.cli.CommandLine;
024import org.apache.commons.cli.Option;
025import org.apache.commons.cli.Options;
026import org.apache.commons.logging.Log;
027import org.apache.commons.logging.LogFactory;
028import org.nuxeo.lib.stream.computation.Record;
029import org.nuxeo.lib.stream.log.LogAppender;
030import org.nuxeo.lib.stream.log.LogManager;
031import org.nuxeo.lib.stream.log.LogRecord;
032import org.nuxeo.lib.stream.log.LogTailer;
033
034/**
035 * Copy a Log to another
036 *
037 * @since 9.3
038 */
039public class CopyCommand extends Command {
040    private static final Log log = LogFactory.getLog(CopyCommand.class);
041
042    protected static final String NAME = "copy";
043
044    @Override
045    public String name() {
046        return NAME;
047    }
048
049    @Override
050    public void updateOptions(Options options) {
051        options.addOption(Option.builder()
052                                .longOpt("src")
053                                .desc("Source log name")
054                                .required()
055                                .hasArg()
056                                .argName("LOG_NAME")
057                                .build());
058        options.addOption(Option.builder()
059                                .longOpt("srcCodec")
060                                .desc("Codec used to read record, can be: java, avro, avroBinary, avroJson")
061                                .hasArg()
062                                .argName("CODEC")
063                                .build());
064        options.addOption(Option.builder()
065                                .longOpt("dest")
066                                .desc("Target log name")
067                                .required()
068                                .hasArg()
069                                .argName("LOG_NAME")
070                                .build());
071        options.addOption(Option.builder()
072                                .longOpt("destCodec")
073                                .desc("Codec used to write record, can be: java, avro, avroBinary, avroJson")
074                                .hasArg()
075                                .argName("CODEC")
076                                .build());
077        options.addOption(Option.builder("g")
078                                .longOpt("group")
079                                .desc("Source consumer group to use")
080                                .hasArg()
081                                .argName("GROUP")
082                                .build());
083    }
084
085    @Override
086    public boolean run(LogManager manager, CommandLine cmd) {
087        String src = cmd.getOptionValue("src");
088        String dest = cmd.getOptionValue("dest");
089        String group = cmd.getOptionValue("group", "tools");
090        String srcCodec = cmd.getOptionValue("srcCodec");
091        String destCodec = cmd.getOptionValue("destCodec");
092        return copy(manager, src, srcCodec, dest, destCodec, group);
093    }
094
095    protected boolean copy(LogManager manager, String src, String srcCodec, String dest, String destCodec,
096            String group) {
097        log.info(String.format("# Copy %s to %s", src, dest));
098        if (!manager.exists(src)) {
099            log.error("source log not found: " + src);
100            return false;
101        }
102        if (manager.exists(dest)) {
103            log.error("destination log already exists: " + dest);
104            return false;
105        }
106        manager.createIfNotExists(dest, manager.size(src));
107        LogAppender<Record> appender = manager.getAppender(dest, getRecordCodec(destCodec));
108        try (LogTailer<Record> tailer = manager.createTailer(group, src, getRecordCodec(srcCodec))) {
109            while (true) {
110                LogRecord<Record> record = tailer.read(Duration.ofSeconds(5));
111                if (record == null) {
112                    break;
113                }
114                appender.append(record.message().getKey(), record.message());
115            }
116        } catch (InterruptedException e) {
117            Thread.currentThread().interrupt();
118            log.error("Interrupted");
119            return false;
120        }
121        return true;
122    }
123
124}