import { Span , SpanStatus } from '@sentry/tracing' ; import { PromiseReadable } from 'promise-readable' ; import { REMOTEDB_MAGIC } from 'src/remotedb/constants' ; import { Binary , Field , FieldType , readField , UInt8 , UInt16 , UInt32 , } from 'src/remotedb/fields' ; import { responseTransform } from 'src/remotedb/message/response' ; import { getMessageName , MessageType , Response } from 'src/remotedb/message/types' ; /**
 * Argument types are used in argument list fields. This is essentially
 * duplicating the field type, but has different values for whatever reason.
 *
 * There do not appear to be argument types for UInt8 and UInt16. At least, no
 * messages include these field types as arguments as far as we know.
 */ enum ArgumentType { String = 0x02 , Binary = 0x03 , UInt32 = 0x06 } /**
 * The message argument list always containts 12 slots
 */ const ARG_COUNT = 12 ; const fieldArgsMap = { [ FieldType . UInt32 ] : ArgumentType . UInt32 , [ FieldType . String ] : ArgumentType . String , [ FieldType . Binary ] : ArgumentType . Binary , [ FieldType . UInt8 ] : 0x00 , [ FieldType . UInt16 ] : 0x00 , } ; const argsFieldMap = { [ ArgumentType . UInt32 ] : FieldType . UInt32 , [ ArgumentType . String ] : FieldType . String , [ ArgumentType . Binary ] : FieldType . Binary , } ; type Options < T extends MessageType > = { transactionId ? : number ; type : T ; args : Field [ ] ; } ; type ResponseType < T > = T extends Response ? T : never ; type Data < T > = ReturnType < typeof responseTransform [ ResponseType < T > ] > ; /**
 * Representation of a set of fields sequenced into a known message format.
 */ export class Message < T extends MessageType = MessageType > { /**
     * Read a single mesasge via a readable stream
     */ static async fromStream < T extends Response > ( stream , expect , span ) { const tx = span . startChild ( { op : 'readFromStream' , description : getMessageName ( expect ) , } ) ; const magicHeader = await readField ( stream , FieldType . UInt32 ) ; if ( magicHeader . value !== REMOTEDB_MAGIC ) { throw new Error ( 'Did not recieve expected magic value. Corrupt message' ) ; } const txId = await readField ( stream , FieldType . UInt32 ) ; const messageType = await readField ( stream , FieldType . UInt16 ) ; const argCount = await readField ( stream , FieldType . UInt8 ) ; const argList = await readField ( stream , FieldType . Binary ) ; const args = new Array ( argCount . value ) ; for ( let i = 0 ; i < argCount . value ; ++ i ) { if ( argList . value [ i ] === ArgumentType . Binary && args [ i - 1 ] ?. value === 0 ) { args [ i ] = new Binary ( Buffer . alloc ( 0 ) ) ; continue ; } args [ i ] = await readField ( stream , argsFieldMap [ argList . value [ i ] as ArgumentType ] ) ; } if ( messageType . value !== expect ) { const expected = expect . toString ( 16 ) ; const actual = messageType . value . toString ( 16 ) ; tx . setStatus ( SpanStatus . FailedPrecondition ) ; tx . finish ( ) ; throw new Error ( `Expected message type 0x${ expected }, got 0x${ actual }` ) ; } tx . finish ( ) ; return new Message ( { transactionId : txId . value , type : messageType . value as T , args , } ) ; } /**
     * The transaction ID is used to associate responses to their requests.
     */ transactionId ? : number ; readonly type : T ; readonly args : Field [ ] ; constructor ( { transactionId , type , args } ) { this . transactionId = transactionId ; this . type = type ; this . args = args ; } /**
     * The byte serialization of the message
     */ get buffer ( ) { const argList = Buffer . alloc ( ARG_COUNT , 0x00 ) ; argList . set ( this . args . map ( arg => fieldArgsMap [ arg . constructor . type ] ) ) ; const args = this . args . reduce < Field [ ] > ( ( args , arg , i ) => { const prevArg = this . args [ i - 1 ] ; const isEmptyBuffer = arg . constructor . type === FieldType . Binary && i !== 0 && prevArg . constructor . type === FieldType . UInt32 && prevArg . value === 0 ; return isEmptyBuffer ? args : [ ... args , arg ] ; } , [ ] ) ; const fields = [ new UInt32 ( REMOTEDB_MAGIC ) , new UInt32 ( this . transactionId ?? 0 ) , new UInt16 ( this . type ) , new UInt8 ( this . args . length ) , new Binary ( argList ) , ... args , ] ; return Buffer . concat ( fields . map ( f => f . buffer ) ) ; } /**
     * The JS representation of the message.
     *
     * Currently only supports representing response messages.
     */ get data ( ) : Data < T > { const type = this . type as ResponseType < T > ; if ( ! Object . values ( Response ) . includes ( type ) ) { throw new Error ( 'Representation of non-responses is not currently supported' ) ; } return responseTransform [ type ] ( this . args ) as Data < T > ; } }