content: /*
   * Copyright (C) 2020 ActiveJ LLC.
   *
   * Licensed 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 io.activej.launchers.crdt;
  
  import io.activej.async.service.ReactiveService;
  import io.activej.crdt.storage.local.FileSystemCrdtStorage;
  import io.activej.crdt.storage.local.MapCrdtStorage;
  import io.activej.datastream.consumer.StreamConsumers;
  import io.activej.promise.Promise;
  import io.activej.reactor.AbstractReactive;
  import io.activej.reactor.Reactor;
  import org.jetbrains.annotations.Nullable;
  
  import static io.activej.reactor.Reactive.checkInReactorThread;
  
  public final class BackupService<K extends Comparable<K>, S> extends AbstractReactive implements ReactiveService {
  	private final MapCrdtStorage<K, S> inMemory;
  	private final FileSystemCrdtStorage<K, S> localFiles;
  
  	private long lastTimestamp = 0;
  
  	private @Nullable Promise<Void> backupPromise = null;
  
  	public BackupService(Reactor reactor, MapCrdtStorage<K, S> inMemory, FileSystemCrdtStorage<K, S> localFiles) {
  		super(reactor);
  		this.inMemory = inMemory;
  		this.localFiles = localFiles;
  	}
  
  	public Promise<Void> restore() {
  		checkInReactorThread(this);
  		return localFiles.download()
  			.then(supplierWithResult ->
  				supplierWithResult.streamTo(StreamConsumers.ofPromise(inMemory.upload())));
  	}
  
  	public Promise<Void> backup() {
  		checkInReactorThread(this);
  		if (backupPromise != null) {
  			return backupPromise;
  		}
  		long lastTimestamp = this.lastTimestamp;
  		this.lastTimestamp = reactor.currentTimeMillis();
  		return backupPromise = inMemory.download(lastTimestamp)
  			.then(supplierWithResult -> supplierWithResult
  				.streamTo(StreamConsumers.ofPromise(localFiles.upload()))
  				.whenComplete(() -> backupPromise = null));
  	}
  
  	public boolean backupInProgress() {
  		return backupPromise != null;
  	}
  
  	@Override
  	public Promise<?> start() {
  		checkInReactorThread(this);
  		return restore().then(localFiles::consolidate);
  	}
  
  	@Override
  	public Promise<?> stop() {
  		checkInReactorThread(this);
  		return backup();
  	}
  }
  