/**
 * Copyright (c) 2014 Jimb Esser
 * Released under the MIT license
 * http://opensource.org/licenses/mit-license.php
 */ /**
 * このソースは https://github.com/Jimbly/node-tail-stream
 * を改変し作成しています。
 */ import * as assert from 'assert' ; import * as fs from 'fs' ; import { Readable , ReadableOptions } from 'stream' ; import ILogger from '../model/ILogger' ; import ILoggerModel from '../model/ILoggerModel' ; import container from '../model/ModelContainer' ; export interface TailStreamOption extends ReadableOptions { start ? : number ; } class TailStream extends Readable { private offset : number ; private closed : boolean = false ; private filePath : string ; private readInProgress : boolean = false ; private getFdInProgress : boolean = false ; private checkIdleTimer : NodeJS . Timeout | null = null ; private checkFileTimer : NodeJS . Timeout | null = null ; private readPending : number = 0 ; private fd : number = 0 ; private log : ILogger ; constructor ( filename , option ) { super ( option ) ; this . filePath = filename ; this . offset = option . start || 0 ; this . log = container . get < ILoggerModel > ( 'ILoggerModel' ) . getLogger ( ) ; this . getFd ( ) ; } private getFd ( ) { assert . ok ( ! this . fd ) ; assert . ok ( ! this . getFdInProgress ) ; this . getFdInProgress = true ; fs . open ( this . filePath , 'r' , ( err , fd ) => { if ( this . closed ) { return ; } assert . ok ( this . getFdInProgress ) ; this . getFdInProgress = false ; if ( err ) { if ( this . readPending ) { setTimeout ( ( ) => { if ( this . closed ) { return ; } this . getFd ( ) ; } , 1000 ) ; } } else { this . fd = fd ; if ( this . readPending ) { this . doRead ( ) ; } } } ) ; } private doRead ( ) { assert . ok ( this . fd ) ; assert . ok ( this . readPending ) ; assert . ok ( ! this . readInProgress ) ; this . readInProgress = true ; fs . fstat ( this . fd , ( err , stat ) => { if ( this . closed ) { return ; } assert . ok ( this . readInProgress ) ; if ( err ) { this . readInProgress = false ; this . debug ( 'error statting' , err ) ; this . emit ( 'error' , err ) ; return ; } let start = this . offset ; const end = stat . size ; if ( end < start ) { start = 0 ; } assert . ok ( this . readPending ) ; const size = Math . min ( this . readPending , end - start ) ; if ( size === 0 ) { this . debug ( 'no data to read' ) ; this . readInProgress = false ; this . checkFile ( ) ; return ; } const buffer = new Buffer ( size ) ; fs . read ( this . fd , buffer , 0 , size , start , ( e , bytesRead , buff ) => { if ( this . closed ) { return ; } assert . ok ( this . readInProgress ) ; this . readInProgress = false ; if ( e ) { this . debug ( 'error reading' , e ) ; return this . emit ( 'error' , e ) ; } if ( bytesRead === 0 ) { this . debug ( 'no data read' ) ; this . checkFile ( ) ; return ; } this . debug ( 'read ' + bytesRead + ' bytes' ) ; this . readPending = 0 ; this . offset = start + bytesRead ; assert . ok ( ! this . checkIdleTimer ) ; this . checkIdleTimer = setTimeout ( ( ) => { this . checkIdle ( ) ; } , 1000 ) ; this . push ( buff ) ; return ; } ) ; } ) ; } private checkFile ( ) { if ( this . checkFileTimer !== null ) { return ; } const stat = fs . statSync ( this . filePath ) ; this . checkFileTimer = setTimeout ( ( ) => { this . checkFileTimer = null ; const newStat = fs . statSync ( this . filePath ) ; if ( newStat . size !== stat . size ) { this . doRead ( ) ; } else { this . close ( ) ; } } , 1000 ) ; } private checkIdle ( ) { assert . ok ( this . checkIdleTimer ) ; this . checkIdleTimer = null ; assert . ok ( ! this . readInProgress && ! this . getFdInProgress ) ; this . debug ( 'timeout expired, closing watcher' ) ; } public _read ( size ) { assert . ok ( ! this . closed ) ; assert . ok ( ! this . readPending ) ; assert . ok ( size ) ; if ( this . checkIdleTimer ) { clearTimeout ( this . checkIdleTimer ) ; this . checkIdleTimer = null ; } this . debug ( 'read_pending = ' + size ) ; this . readPending = size ; if ( ! this . fd ) { if ( this . getFdInProgress ) { this . debug ( 'waiting on fd' ) ; } else { this . getFd ( ) ; } return ; } this . doRead ( ) ; } private close ( ) { this . closed = true ; this . push ( null ) ; } private debug ( str , err ? ) { if ( err ) { this . log . stream . error ( str ) ; this . log . stream . error ( err ) ; } else { } } } export const createReadStream = ( path , option ) : TailStream => { return new TailStream ( path , option ) ; } ;