|
|
@@ -0,0 +1,470 @@
|
|
|
+import Logger from '../util/Logger';
|
|
|
+import { fileIo } from '@kit.CoreFileKit';
|
|
|
+import { BusinessError } from '@kit.BasicServicesKit';
|
|
|
+import { httpServer, HttpRequest, HttpResponse } from '@webabcd/harmony-httpserver';
|
|
|
+import { readSmbRange, SmbConnectionInfo } from './SmbRangeReader';
|
|
|
+import { readFtpRange, FtpConnectionInfo } from './FtpRangeReader';
|
|
|
+
|
|
|
+const TAG = 'SmbStreamingHttpServer';
|
|
|
+const STREAM_ROUTE_PREFIX = '/smb-stream';
|
|
|
+const LOCAL_HOST = '127.0.0.1';
|
|
|
+const MAX_CHUNK_SIZE = 512 * 1024; // 512KB per response
|
|
|
+const RANGE_WAIT_TIMEOUT_MS = 60000;
|
|
|
+const RANGE_POLL_INTERVAL_MS = 100;
|
|
|
+const SESSION_TTL_MS = 10 * 60 * 1000;
|
|
|
+const PORT_CANDIDATES: number[] = [18888, 18889, 18900, 18901, 18902];
|
|
|
+
|
|
|
+type StreamingProtocol = 'smb' | 'ftp';
|
|
|
+
|
|
|
+interface StreamingSessionOptions {
|
|
|
+ sessionKey: string;
|
|
|
+ cachePath: string;
|
|
|
+ expectedSize?: number;
|
|
|
+ mimeType?: string;
|
|
|
+ fileName?: string;
|
|
|
+ remotePath?: string;
|
|
|
+ protocol?: StreamingProtocol;
|
|
|
+ smbConnection?: SmbConnectionInfo;
|
|
|
+ ftpConnection?: FtpConnectionInfo;
|
|
|
+}
|
|
|
+
|
|
|
+interface StreamingSession extends StreamingSessionOptions {
|
|
|
+ lastAccess: number;
|
|
|
+}
|
|
|
+
|
|
|
+interface RangeInfo {
|
|
|
+ start: number;
|
|
|
+ end?: number;
|
|
|
+ isRange: boolean;
|
|
|
+}
|
|
|
+
|
|
|
+interface ChunkResult {
|
|
|
+ buffer: ArrayBuffer;
|
|
|
+ start: number;
|
|
|
+ end: number;
|
|
|
+ total?: number;
|
|
|
+}
|
|
|
+
|
|
|
+class RangeNotSatisfiableError extends Error {
|
|
|
+ availableRange: string;
|
|
|
+
|
|
|
+ constructor(availableRange: string) {
|
|
|
+ super('Requested Range Not Satisfiable');
|
|
|
+ this.name = 'RangeNotSatisfiableError';
|
|
|
+ this.availableRange = availableRange;
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+export default class SmbStreamingHttpServer {
|
|
|
+ private static instance?: SmbStreamingHttpServer;
|
|
|
+ private port: number = -1;
|
|
|
+ private sessions: Map<string, StreamingSession> = new Map();
|
|
|
+ private startPromise?: Promise<void>;
|
|
|
+
|
|
|
+ static getInstance(): SmbStreamingHttpServer {
|
|
|
+ if (!SmbStreamingHttpServer.instance) {
|
|
|
+ SmbStreamingHttpServer.instance = new SmbStreamingHttpServer();
|
|
|
+ }
|
|
|
+ return SmbStreamingHttpServer.instance;
|
|
|
+ }
|
|
|
+
|
|
|
+ async getStreamingUrl(options: StreamingSessionOptions): Promise<string> {
|
|
|
+ await this.ensureServerStarted();
|
|
|
+ const now = Date.now();
|
|
|
+ const existing = this.sessions.get(options.sessionKey);
|
|
|
+ if (existing) {
|
|
|
+ existing.lastAccess = now;
|
|
|
+ existing.cachePath = options.cachePath;
|
|
|
+ existing.expectedSize = options.expectedSize;
|
|
|
+ existing.mimeType = options.mimeType;
|
|
|
+ existing.fileName = options.fileName;
|
|
|
+ existing.remotePath = options.remotePath;
|
|
|
+ existing.protocol = options.protocol;
|
|
|
+ existing.smbConnection = options.smbConnection;
|
|
|
+ existing.ftpConnection = options.ftpConnection;
|
|
|
+ Logger.debug(TAG, `更新SMB流会话: ${options.sessionKey}`);
|
|
|
+ } else {
|
|
|
+ this.sessions.set(options.sessionKey, {
|
|
|
+ sessionKey: options.sessionKey,
|
|
|
+ cachePath: options.cachePath,
|
|
|
+ expectedSize: options.expectedSize,
|
|
|
+ mimeType: options.mimeType,
|
|
|
+ fileName: options.fileName,
|
|
|
+ remotePath: options.remotePath,
|
|
|
+ protocol: options.protocol,
|
|
|
+ smbConnection: options.smbConnection,
|
|
|
+ ftpConnection: options.ftpConnection,
|
|
|
+ lastAccess: now
|
|
|
+ });
|
|
|
+ Logger.info(TAG, `注册SMB流会话: ${options.sessionKey}`);
|
|
|
+ }
|
|
|
+ this.cleanupExpiredSessions();
|
|
|
+ return `http://${LOCAL_HOST}:${this.port}${STREAM_ROUTE_PREFIX}/${encodeURIComponent(options.sessionKey)}`;
|
|
|
+ }
|
|
|
+
|
|
|
+ private async ensureServerStarted(): Promise<void> {
|
|
|
+ if (this.port > 0) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (this.startPromise) {
|
|
|
+ return this.startPromise;
|
|
|
+ }
|
|
|
+ this.startPromise = this.startServerInternal().finally(() => {
|
|
|
+ this.startPromise = undefined;
|
|
|
+ });
|
|
|
+ return this.startPromise;
|
|
|
+ }
|
|
|
+
|
|
|
+ private async startServerInternal(): Promise<void> {
|
|
|
+ httpServer.enableLog(false);
|
|
|
+ this.port = await this.tryStartOnAvailablePort();
|
|
|
+ httpServer.handleHttpRequestAsync((request: HttpRequest) => this.handleRequest(request));
|
|
|
+ Logger.info(TAG, `SMB流HTTP服务启动,端口: ${this.port}`);
|
|
|
+ }
|
|
|
+
|
|
|
+ private async tryStartOnAvailablePort(): Promise<number> {
|
|
|
+ for (let i = 0; i < PORT_CANDIDATES.length; i++) {
|
|
|
+ const candidate = PORT_CANDIDATES[i];
|
|
|
+ try {
|
|
|
+ return await this.startOnPort(candidate);
|
|
|
+ } catch (error) {
|
|
|
+ const err = error as Error;
|
|
|
+ Logger.warn(TAG, `端口${candidate}启动失败: ${err.message}`);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ throw new Error('无法启动SMB流HTTP服务,端口均不可用');
|
|
|
+ }
|
|
|
+
|
|
|
+ private startOnPort(port: number): Promise<number> {
|
|
|
+ return new Promise((resolve, reject) => {
|
|
|
+ try {
|
|
|
+ httpServer.start(port, (error: BusinessError, realPort: number) => {
|
|
|
+ if (error && error.code !== 0) {
|
|
|
+ reject(new Error(`HTTP服务启动失败:${error.code}-${error.message}`));
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ resolve(realPort);
|
|
|
+ });
|
|
|
+ } catch (error) {
|
|
|
+ reject(error as Error);
|
|
|
+ }
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ private async handleRequest(request: HttpRequest): Promise<HttpResponse> {
|
|
|
+ try {
|
|
|
+ this.cleanupExpiredSessions();
|
|
|
+ if (!request || !request.url) {
|
|
|
+ return this.buildPlainResponse(400, 'invalid request');
|
|
|
+ }
|
|
|
+ const sessionId = this.extractSessionId(request.url);
|
|
|
+ Logger.debug(TAG, `收到请求 method=${request.method} url=${request.url} range=${this.getHeader(request.headers, 'range') ?? ''}`);
|
|
|
+ if (!sessionId) {
|
|
|
+ return this.buildPlainResponse(404, 'not found');
|
|
|
+ }
|
|
|
+ const session = this.sessions.get(sessionId);
|
|
|
+ if (!session) {
|
|
|
+ return this.buildPlainResponse(404, 'session expired');
|
|
|
+ }
|
|
|
+ session.lastAccess = Date.now();
|
|
|
+ const method = (request.method || 'GET').toUpperCase();
|
|
|
+ if (method !== 'GET' && method !== 'HEAD') {
|
|
|
+ return this.buildPlainResponse(405, 'method not allowed');
|
|
|
+ }
|
|
|
+ if (method === 'HEAD') {
|
|
|
+ return this.buildHeadResponse(session);
|
|
|
+ }
|
|
|
+
|
|
|
+ const rangeHeader = this.getHeader(request.headers, 'range');
|
|
|
+ const rangeInfo = this.parseRangeHeader(rangeHeader);
|
|
|
+ let chunk: ChunkResult;
|
|
|
+ try {
|
|
|
+ chunk = await this.readChunk(session, rangeInfo.start, rangeInfo.end);
|
|
|
+ } catch (error) {
|
|
|
+ const err = error as Error;
|
|
|
+ if (err instanceof RangeNotSatisfiableError) {
|
|
|
+ return this.buildRangeNotSatisfiableResponse(err.availableRange);
|
|
|
+ }
|
|
|
+ throw err;
|
|
|
+ }
|
|
|
+ return this.buildChunkResponse(session, chunk, rangeInfo.isRange);
|
|
|
+ } catch (error) {
|
|
|
+ const err = error as Error;
|
|
|
+ Logger.error(TAG, `处理SMB流请求失败: ${err.message}`);
|
|
|
+ return this.buildPlainResponse(500, err.message || 'internal error');
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private buildPlainResponse(statusCode: number, message: string): HttpResponse {
|
|
|
+ return {
|
|
|
+ statusCode,
|
|
|
+ result: message,
|
|
|
+ headers: {
|
|
|
+ 'Content-Type': 'text/plain; charset=utf-8'
|
|
|
+ }
|
|
|
+ } as HttpResponse;
|
|
|
+ }
|
|
|
+
|
|
|
+ private buildRangeNotSatisfiableResponse(availableRange: string): HttpResponse {
|
|
|
+ return {
|
|
|
+ statusCode: 416,
|
|
|
+ result: 'Requested Range Not Satisfiable',
|
|
|
+ headers: {
|
|
|
+ 'Content-Range': availableRange,
|
|
|
+ 'Content-Type': 'text/plain; charset=utf-8'
|
|
|
+ }
|
|
|
+ } as HttpResponse;
|
|
|
+ }
|
|
|
+
|
|
|
+ private buildHeadResponse(session: StreamingSession): HttpResponse {
|
|
|
+ const headers: Record<string, string> = {
|
|
|
+ 'Accept-Ranges': 'bytes',
|
|
|
+ 'Content-Type': session.mimeType || 'application/octet-stream'
|
|
|
+ };
|
|
|
+ if (session.expectedSize && session.expectedSize > 0) {
|
|
|
+ headers['Content-Length'] = session.expectedSize.toString();
|
|
|
+ }
|
|
|
+ return {
|
|
|
+ statusCode: 200,
|
|
|
+ headers,
|
|
|
+ result: ''
|
|
|
+ } as HttpResponse;
|
|
|
+ }
|
|
|
+
|
|
|
+ private buildChunkResponse(session: StreamingSession, chunk: ChunkResult, isRange: boolean): HttpResponse {
|
|
|
+ const headers: Record<string, string> = {
|
|
|
+ 'Content-Type': session.mimeType || 'application/octet-stream',
|
|
|
+ 'Accept-Ranges': 'bytes',
|
|
|
+ 'Content-Length': (chunk.end - chunk.start + 1).toString()
|
|
|
+ };
|
|
|
+ if (chunk.total && chunk.total > 0) {
|
|
|
+ headers['Content-Range'] = `bytes ${chunk.start}-${chunk.end}/${chunk.total}`;
|
|
|
+ }
|
|
|
+ return {
|
|
|
+ statusCode: isRange ? 206 : 200,
|
|
|
+ headers,
|
|
|
+ result: chunk.buffer
|
|
|
+ } as HttpResponse;
|
|
|
+ }
|
|
|
+
|
|
|
+ private extractSessionId(url: string): string | undefined {
|
|
|
+ if (!url) {
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ if (url.startsWith('http://') || url.startsWith('https://')) {
|
|
|
+ const slashIndex = url.indexOf('/', url.indexOf('//') + 2);
|
|
|
+ url = slashIndex >= 0 ? url.substring(slashIndex) : '/';
|
|
|
+ }
|
|
|
+ if (!url.startsWith(STREAM_ROUTE_PREFIX)) {
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ let relative = url.substring(STREAM_ROUTE_PREFIX.length);
|
|
|
+ if (relative.startsWith('/')) {
|
|
|
+ relative = relative.substring(1);
|
|
|
+ }
|
|
|
+ const queryIndex = relative.indexOf('?');
|
|
|
+ if (queryIndex >= 0) {
|
|
|
+ relative = relative.substring(0, queryIndex);
|
|
|
+ }
|
|
|
+ if (!relative || relative.length === 0) {
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ return decodeURIComponent(relative);
|
|
|
+ } catch (error) {
|
|
|
+ Logger.error(TAG, `sessionId解析失败: ${(error as Error).message}`);
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private parseRangeHeader(header?: string): RangeInfo {
|
|
|
+ if (!header || header.length === 0) {
|
|
|
+ return { start: 0, isRange: false };
|
|
|
+ }
|
|
|
+ const match = header.match(/bytes=([0-9]*)-([0-9]*)/i);
|
|
|
+ if (!match) {
|
|
|
+ return { start: 0, isRange: false };
|
|
|
+ }
|
|
|
+ let start = match[1] ? parseInt(match[1]) : 0;
|
|
|
+ const hasEnd = match[2] && match[2].length > 0;
|
|
|
+ let end = hasEnd ? parseInt(match[2]) : undefined;
|
|
|
+ if (isNaN(start) || start < 0) {
|
|
|
+ start = 0;
|
|
|
+ }
|
|
|
+ if (end === undefined || isNaN(end) || end < start) {
|
|
|
+ end = start + MAX_CHUNK_SIZE - 1;
|
|
|
+ }
|
|
|
+ const prefixRange = match[1] === undefined || match[1].length === 0;
|
|
|
+ if (prefixRange) {
|
|
|
+ const suffixBytes = parseInt(match[2]);
|
|
|
+ if (!isNaN(suffixBytes) && suffixBytes > 0) {
|
|
|
+ start = Math.max(0, end - suffixBytes + 1);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (end !== undefined && (isNaN(end) || end < start)) {
|
|
|
+ end = start;
|
|
|
+ }
|
|
|
+ return { start, end, isRange: true };
|
|
|
+ }
|
|
|
+
|
|
|
+ private async readChunk(session: StreamingSession, requestedStart: number, requestedEnd?: number): Promise<ChunkResult> {
|
|
|
+ const start = requestedStart >= 0 ? requestedStart : 0;
|
|
|
+ let end = requestedEnd !== undefined ? requestedEnd : start + MAX_CHUNK_SIZE - 1;
|
|
|
+ if (end - start + 1 > MAX_CHUNK_SIZE) {
|
|
|
+ end = start + MAX_CHUNK_SIZE - 1;
|
|
|
+ }
|
|
|
+
|
|
|
+ const isFtp = session.protocol === 'ftp';
|
|
|
+ const timeout = isFtp ? RANGE_WAIT_TIMEOUT_MS * 3 : RANGE_WAIT_TIMEOUT_MS;
|
|
|
+ const pollInterval = isFtp ? RANGE_POLL_INTERVAL_MS * 2 : RANGE_POLL_INTERVAL_MS;
|
|
|
+
|
|
|
+ const waitStart = Date.now();
|
|
|
+ while (Date.now() - waitStart <= timeout) {
|
|
|
+ const size = await this.tryGetFileSize(session.cachePath);
|
|
|
+ Logger.debug(TAG, `range waiting start=${start} end=${end} currentSize=${size} cachePath=${session.cachePath} protocol=${session.protocol || 'unknown'}`);
|
|
|
+
|
|
|
+ if (size >= 0 && size > start) {
|
|
|
+ let fileEnd = Math.min(end, size - 1);
|
|
|
+ if (session.expectedSize && session.expectedSize > 0) {
|
|
|
+ fileEnd = Math.min(fileEnd, session.expectedSize - 1);
|
|
|
+ }
|
|
|
+ if (fileEnd >= start) {
|
|
|
+ const length = fileEnd - start + 1;
|
|
|
+ const buffer = new ArrayBuffer(length);
|
|
|
+ const file = fileIo.openSync(session.cachePath, fileIo.OpenMode.READ_ONLY);
|
|
|
+ try {
|
|
|
+ fileIo.readSync(file.fd, buffer, { offset: start, length });
|
|
|
+ } finally {
|
|
|
+ fileIo.closeSync(file);
|
|
|
+ }
|
|
|
+ const total = session.expectedSize && session.expectedSize > 0 ? session.expectedSize : Math.max(size, fileEnd + 1);
|
|
|
+ Logger.info(TAG, `从本地缓存返回范围 ${start}-${fileEnd},大小: ${length} bytes`);
|
|
|
+ return {
|
|
|
+ buffer,
|
|
|
+ start,
|
|
|
+ end: fileEnd,
|
|
|
+ total
|
|
|
+ };
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ const hasLocalData = size > 0;
|
|
|
+ const requestExceedsLocal = start >= size;
|
|
|
+
|
|
|
+ if (requestExceedsLocal) {
|
|
|
+ if (session.protocol === 'ftp' && !hasLocalData) {
|
|
|
+ Logger.debug(TAG, 'FTP本地无数据,直接尝试远程读取');
|
|
|
+ }
|
|
|
+ const remoteChunk = await this.fetchRemoteChunk(session, start, end, size);
|
|
|
+ if (remoteChunk && remoteChunk.buffer.byteLength > 0) {
|
|
|
+ Logger.info(TAG, `远程读取成功,返回 ${remoteChunk.buffer.byteLength} bytes,实现边下边播`);
|
|
|
+ return remoteChunk;
|
|
|
+ }
|
|
|
+
|
|
|
+ if (hasLocalData) {
|
|
|
+ Logger.info(TAG, `用户请求未下载区域 ${start}-${end},已下载到 ${size - 1},返回HTTP 416让播放器跳回`);
|
|
|
+ throw new RangeNotSatisfiableError(`bytes 0-${size - 1}/${session.expectedSize || '*'}`);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if (session.expectedSize && session.expectedSize > 0 && start >= session.expectedSize) {
|
|
|
+ throw new Error('请求范围超出文件大小');
|
|
|
+ }
|
|
|
+ await this.delay(pollInterval);
|
|
|
+ }
|
|
|
+ Logger.error(TAG, `range等待超时 start=${start} end=${end} cache=${session.cachePath} protocol=${session.protocol || 'unknown'}`);
|
|
|
+ throw new Error(`等待${session.protocol === 'ftp' ? 'FTP' : 'SMB'}缓存数据超时`);
|
|
|
+ }
|
|
|
+
|
|
|
+ private async tryGetFileSize(path: string): Promise<number> {
|
|
|
+ try {
|
|
|
+ const stat = await fileIo.stat(path);
|
|
|
+ return stat.size;
|
|
|
+ } catch (error) {
|
|
|
+ return -1;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private getHeader(headers: Record<string, string> | undefined, name: string): string | undefined {
|
|
|
+ if (!headers) {
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ const target = name.toLowerCase();
|
|
|
+ const keys = Object.keys(headers);
|
|
|
+ for (let i = 0; i < keys.length; i++) {
|
|
|
+ const key = keys[i];
|
|
|
+ if (key.toLowerCase() === target) {
|
|
|
+ return headers[key];
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+
|
|
|
+ private cleanupExpiredSessions(): void {
|
|
|
+ const now = Date.now();
|
|
|
+ const keys = Array.from(this.sessions.keys());
|
|
|
+ for (let i = 0; i < keys.length; i++) {
|
|
|
+ const key = keys[i];
|
|
|
+ const session = this.sessions.get(key);
|
|
|
+ if (session && now - session.lastAccess > SESSION_TTL_MS) {
|
|
|
+ this.sessions.delete(key);
|
|
|
+ Logger.info(TAG, `移除过期SMB流会话: ${key}`);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private delay(ms: number): Promise<void> {
|
|
|
+ return new Promise(resolve => setTimeout(resolve, ms));
|
|
|
+ }
|
|
|
+
|
|
|
+ private async fetchRemoteChunk(
|
|
|
+ session: StreamingSession,
|
|
|
+ start: number,
|
|
|
+ desiredEnd: number,
|
|
|
+ currentLocalSize: number
|
|
|
+ ): Promise<ChunkResult | undefined> {
|
|
|
+ if (!session.protocol || !session.remotePath) {
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ const length = desiredEnd - start + 1;
|
|
|
+ if (length <= 0) {
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ let data: ArrayBuffer | undefined;
|
|
|
+ if (session.protocol === 'smb' && session.smbConnection) {
|
|
|
+ data = readSmbRange(session.smbConnection, session.remotePath, start, length);
|
|
|
+ } else if (session.protocol === 'ftp' && session.ftpConnection) {
|
|
|
+ data = await readFtpRange(session.ftpConnection, session.remotePath, start, length);
|
|
|
+ }
|
|
|
+ if (!data || data.byteLength === 0) {
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ const shouldWriteToCache = currentLocalSize >= 0 && start <= currentLocalSize;
|
|
|
+ if (shouldWriteToCache) {
|
|
|
+ await this.writeRangeToCache(session.cachePath, start, data);
|
|
|
+ } else {
|
|
|
+ Logger.debug(TAG, `跳过写入缓存,避免产生空洞: start=${start}, currentLocalSize=${currentLocalSize}`);
|
|
|
+ }
|
|
|
+ const chunkEnd = start + data.byteLength - 1;
|
|
|
+ const totalSize = session.expectedSize && session.expectedSize > 0 ? session.expectedSize : Math.max(chunkEnd + 1, await this.tryGetFileSize(session.cachePath));
|
|
|
+ return {
|
|
|
+ buffer: data,
|
|
|
+ start,
|
|
|
+ end: chunkEnd,
|
|
|
+ total: totalSize
|
|
|
+ };
|
|
|
+ } catch (error) {
|
|
|
+ Logger.error(TAG, `远程读取失败: ${(error as Error).message}`);
|
|
|
+ return undefined;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ private async writeRangeToCache(path: string, offset: number, data: ArrayBuffer): Promise<void> {
|
|
|
+ const file = fileIo.openSync(path, fileIo.OpenMode.CREATE | fileIo.OpenMode.READ_WRITE);
|
|
|
+ try {
|
|
|
+ fileIo.writeSync(file.fd, data, { offset, length: data.byteLength });
|
|
|
+ } finally {
|
|
|
+ fileIo.closeSync(file);
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|