timi-compress/src/main/java/com/imyeyu/compress/AbstractRunner.java

package com.imyeyu.compress;

import com.imyeyu.java.bean.ActionException;
import com.imyeyu.java.bean.CallbackArg;
import com.imyeyu.utils.OS;

import java.io.File;
import java.io.FilterInputStream;
import java.io.FilterOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

/**
 * 抽象处理器
 *
 * @param <T> 处理器类型
 * @author 夜雨
 * @since 2024-06-30 18:09
 */
public abstract class AbstractRunner<T extends AbstractRunner<T>> implements OS.FileSystem {

	/** 文件处理回调 */
	protected CallbackArg<File> fileCallback;

	/** 进度回调 */
	protected CallbackArg<Double> progressCallback;

	/** 中断标记 */
	protected boolean isInterrupt = false;

	/** 暂停标记 */
	protected boolean isPause = false;

	/** 暂停锁 */
	protected final Object pauseLock = new Object();

	/** 总字节数 */
	private long progressTotalBytes = -1L;

	/** 当前已处理字节数 */
	private long progressBytes = 0L;

	/**
	 * 设置文件处理回调
	 *
	 * @param fileCallback 文件处理回调
	 * @return 当前处理器
	 */
	public T setFileCallback(CallbackArg<File> fileCallback) {
		this.fileCallback = fileCallback;
		return self();
	}

	/**
	 * 设置文件处理回调
	 *
	 * @param fileCallback 文件处理回调
	 * @return 当前处理器
	 */
	public T fileCallback(CallbackArg<File> fileCallback) {
		return setFileCallback(fileCallback);
	}

	/**
	 * 设置进度回调
	 *
	 * @param progressCallback 进度回调
	 * @return 当前处理器
	 */
	public T setProgressCallback(CallbackArg<Double> progressCallback) {
		this.progressCallback = progressCallback;
		return self();
	}

	/**
	 * 设置进度回调
	 *
	 * @param progressCallback 进度回调
	 * @return 当前处理器
	 */
	public T progressCallback(CallbackArg<Double> progressCallback) {
		return setProgressCallback(progressCallback);
	}

	/** 暂停任务 */
	public void pause() {
		isPause = true;
	}

	/** 恢复任务 */
	public void start() {
		isPause = false;
		synchronized (pauseLock) {
			pauseLock.notifyAll();
		}
	}

	/** 中断任务 */
	public void interrupt() {
		isInterrupt = true;
	}

	/**
	 * 返回当前处理器
	 *
	 * @return 当前处理器
	 */
	@SuppressWarnings("unchecked")
	protected final T self() {
		return (T) this;
	}

	/**
	 * 初始化字节进度
	 *
	 * @param totalBytes 总字节数
	 */
	protected void initByteProgress(long totalBytes) {
		progressBytes = 0L;
		progressTotalBytes = totalBytes;
	}

	/** 重置进度状态 */
	protected void resetProgress() {
		progressBytes = 0L;
		progressTotalBytes = -1L;
	}

	/**
	 * 执行暂停检查
	 *
	 * @throws InterruptedException 等待恢复时被中断
	 */
	protected void awaitIfPaused() throws InterruptedException {
		if (!isPause) {
			return;
		}
		synchronized (pauseLock) {
			while (isPause) {
				pauseLock.wait();
			}
		}
	}

	/**
	 * 检查是否已经中断
	 *
	 * @throws IOException 操作已中断
	 */
	protected void ensureRunning() throws IOException {
		if (isInterrupt) {
			throw new IOException("操作已中断");
		}
	}

	/**
	 * 触发文件回调
	 *
	 * @param file 当前处理的文件
	 */
	protected void handleFile(File file) {
		if (fileCallback != null) {
			fileCallback.handler(file);
		}
	}

	/**
	 * 触发进度回调
	 *
	 * @param currentBytes 当前已处理字节数
	 * @param totalBytes 总字节数
	 */
	protected void handleProgress(long currentBytes, long totalBytes) {
		if (progressCallback == null || totalBytes < 1L) {
			return;
		}
		double progress = Math.min(1D, currentBytes * 1D / totalBytes);
		progressCallback.handler(progress);
	}

	/**
	 * 增加已处理字节数并回调
	 *
	 * @param bytes 本次新增字节数
	 */
	protected void handleTransferred(long bytes) {
		if (bytes < 1L) {
			return;
		}
		progressBytes += bytes;
		handleProgress(progressBytes, progressTotalBytes);
	}

	/** 将进度推进到完成态 */
	protected void finishProgress() {
		if (progressCallback != null) {
			progressCallback.handler(1D);
		}
	}

	/**
	 * 复制数据流并更新进度
	 *
	 * @param fromStream 输入流
	 * @param toStream 输出流
	 * @return 已复制字节数
	 * @throws Exception 复制失败
	 */
	protected long transfer(InputStream fromStream, OutputStream toStream) throws Exception {
		return transfer(fromStream, toStream, true);
	}

	/**
	 * 复制数据流
	 *
	 * @param fromStream 输入流
	 * @param toStream 输出流
	 * @param countProgress 是否统计进度
	 * @return 已复制字节数
	 * @throws Exception 复制失败
	 */
	protected long transfer(InputStream fromStream, OutputStream toStream, boolean countProgress) throws Exception {
		byte[] buffer = new byte[8192];
		long total = 0L;
		int length;
		while ((length = fromStream.read(buffer)) != -1) {
			awaitIfPaused();
			ensureRunning();
			toStream.write(buffer, 0, length);
			total += length;
			if (countProgress) {
				handleTransferred(length);
			}
		}
		return total;
	}

	/**
	 * 创建一个关闭时不关闭原始流的输入流包装
	 *
	 * @param inputStream 原始输入流
	 * @return 包装后的输入流
	 */
	protected InputStream nonClosing(InputStream inputStream) {
		return new FilterInputStream(inputStream) {

			@Override
			public void close() {
			}
		};
	}

	/**
	 * 创建一个关闭时不关闭原始流的输出流包装
	 *
	 * @param outputStream 原始输出流
	 * @return 包装后的输出流
	 */
	protected OutputStream nonClosing(OutputStream outputStream) {
		return new FilterOutputStream(outputStream) {

			@Override
			public void close() throws IOException {
				flush();
			}
		};
	}

	/**
	 * 规范化归档条目名称
	 *
	 * @param entryName 条目名称
	 * @return 规范化后的名称
	 */
	protected String normalizeEntryName(String entryName) {
		ActionException.required(entryName, "not found entryName");
		String normalized = entryName.replace('\\', '/');
		while (normalized.startsWith("/")) {
			normalized = normalized.substring(1);
		}
		ActionException.required(normalized, "not found normalized entryName");
		return normalized;
	}
}