import debug from 'debug' ; import { byte , util } from 'apache-dubbo-common' ; import { IObservable , TDecodeBuffSubscriber } from './types' ; import { DUBBO_HEADER_LENGTH , DUBBO_MAGIC_HIGH , DUBBO_MAGIC_LOW } from './constants' ; import { Socket } from 'net' ; const log = debug ( 'dubbo:decode-buffer' ) ; /**
 * 在并发的tcp数据传输中，会出现少包，粘包的现象
 * 好在tcp的传输是可以保证顺序的
 * 我们需要抽取一个buffer来统一处理这些数据
 */ export default class DecodeBuffer implements IObservable < TDecodeBuffSubscriber > { private buffer : Buffer ; private transport : Socket ; private subscriber : Function ; constructor ( transport , private flag ) { log ( '%s new DecodeBuffer' , this . flag ) ; this . subscriber = util . noop ; this . buffer = Buffer . alloc ( 0 ) ; this . transport = transport ; process . nextTick ( ( ) => { this . transport . on ( 'data' , ( data ) => this . receive ( data ) ) . on ( 'close' , ( ) => { this . clearBuffer ( ) ; } ) ; } ) ; } static from ( transport , flag ) { return new DecodeBuffer ( transport , flag ) ; } receive ( data ) { this . buffer = Buffer . concat ( [ this . buffer , data ] ) ; let bufferLength = this . buffer . length ; while ( bufferLength >= DUBBO_HEADER_LENGTH ) { const magicHigh = this . buffer [ 0 ] ; const magicLow = this . buffer [ 1 ] ; if ( magicHigh != DUBBO_MAGIC_HIGH || magicLow != DUBBO_MAGIC_LOW ) { log ( `%s receive server data error, buffer[0] is 0xda ${ magicHigh == 0xda } buffer[1] is 0xbb ${ magicLow == 0xbb }` , this . flag ) ; const magicHighIndex = this . buffer . indexOf ( DUBBO_MAGIC_HIGH ) ; const magicLowIndex = this . buffer . indexOf ( DUBBO_MAGIC_LOW ) ; log ( `%s magicHigh index#${ magicHighIndex }` , this . flag ) ; log ( `%s magicLow index#${ magicLowIndex }` , this . flag ) ; if ( magicHighIndex === - 1 ) { this . buffer = this . buffer . slice ( bufferLength ) ; } else if ( magicLowIndex === - 1 ) { if ( magicHighIndex === bufferLength - 1 ) { this . buffer = this . buffer . slice ( magicHighIndex ) ; } else { this . buffer = this . buffer . slice ( bufferLength ) ; } } else { if ( magicLowIndex - magicHighIndex === 1 ) { this . buffer = this . buffer . slice ( magicHighIndex ) ; } else { this . buffer = this . buffer . slice ( magicHighIndex + 1 ) ; } } bufferLength = this . buffer . length ; if ( bufferLength < DUBBO_HEADER_LENGTH ) { return ; } } else { if ( bufferLength < DUBBO_HEADER_LENGTH ) { log ( '%s bufferLength < header length' , this . flag ) ; return ; } const header = this . buffer . slice ( 0 , DUBBO_HEADER_LENGTH ) ; const bodyLengthBuff = Buffer . from ( [ header [ 12 ] , header [ 13 ] , header [ 14 ] , header [ 15 ] ] ) ; const bodyLength = byte . fromBytes4 ( bodyLengthBuff ) ; log ( '%s body length %d' , this . flag , bodyLength ) ; if ( DUBBO_HEADER_LENGTH + bodyLength > bufferLength ) { log ( '%s header length + body length > buffer length' , this . flag ) ; return ; } const dataBuffer = this . buffer . slice ( 0 , DUBBO_HEADER_LENGTH + bodyLength ) ; this . buffer = this . buffer . slice ( DUBBO_HEADER_LENGTH + bodyLength ) ; bufferLength = this . buffer . length ; this . subscriber ( dataBuffer ) ; } } } clearBuffer ( ) { if ( this . buffer . length > 0 ) { this . buffer = Buffer . alloc ( 0 ) ; } } subscribe ( subscriber ) { this . subscriber = subscriber ; return this ; } }