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'; import { connection } from '@kit.NetworkKit'; const TAG = 'SmbStreamingHttpServer'; const STREAM_ROUTE_PREFIX = '/smb-stream'; const DEFAULT_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 = new Map(); private startPromise?: Promise; static getInstance(): SmbStreamingHttpServer { if (!SmbStreamingHttpServer.instance) { SmbStreamingHttpServer.instance = new SmbStreamingHttpServer(); } return SmbStreamingHttpServer.instance; } async getStreamingUrl(options: StreamingSessionOptions): Promise { await this.ensureServerStarted(); const host = await this.resolveLocalHost(); 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://${host}:${this.port}${STREAM_ROUTE_PREFIX}/${encodeURIComponent(options.sessionKey)}`; } private async resolveLocalHost(): Promise { try { const netHandle = await connection.getDefaultNet(); const addressGetter = (connection as unknown as { getAddressesByNetwork?: (handle: unknown) => Promise }) .getAddressesByNetwork; if (!addressGetter) { return DEFAULT_HOST; } const addresses = await addressGetter(netHandle); if (addresses && addresses.length > 0) { for (let i = 0; i < addresses.length; i++) { const raw = addresses[i] as Record; const address = (raw.address ?? raw.addr ?? raw.ip) as string | undefined; if (!address) { continue; } if (address.startsWith('127.') || address === '0.0.0.0') { continue; } if (address.includes(':')) { continue; } return address; } } } catch (error) { Logger.warn(TAG, `解析本机IP失败: ${(error as Error).message}`); } return DEFAULT_HOST; } private async ensureServerStarted(): Promise { 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 { 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 { 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 { 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 { 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 = { '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 = { '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 { 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 { try { const stat = await fileIo.stat(path); return stat.size; } catch (error) { return -1; } } private getHeader(headers: Record | 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 { return new Promise(resolve => setTimeout(resolve, ms)); } private async fetchRemoteChunk( session: StreamingSession, start: number, desiredEnd: number, currentLocalSize: number ): Promise { 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 { 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); } } }