typescript/pxprpc/extend.ts

   1 import { Client, PxpCallable, PxpObject, PxpRequest, Server } from './base'
   2
   3 export class ByteBuffer{
   4     public dv?:DataView;
   5     public pos:number=0;
   6     public constructor(buf:ArrayBuffer){
   7         this.dv=new DataView(buf);
   8     }
   9     public getByte(){
  10         let val=this.dv!.getInt8(this.pos);
  11         this.pos++;
  12         return val;
  13     }
  14     public getInt(){
  15         let val=this.dv!.getInt32(this.pos,true);
  16         this.pos+=4;
  17         return val;
  18     }
  19     public getLong(){
  20         let val=this.dv!.getBigInt64(this.pos,true);
  21         this.pos+=8;
  22         return val;
  23     }
  24     public getFloat(){
  25         let val=this.dv!.getFloat32(this.pos,true);
  26         this.pos+=4;
  27         return val;
  28     }
  29     public getDouble(){
  30         let val=this.dv!.getInt32(this.pos,true);
  31         this.pos+=8;
  32         return val;
  33     }
  34     public getBytes(len:number){
  35         let val=this.dv!.buffer!.slice(this.pos,len);
  36         this.pos+=len;
  37         return val;
  38     }
  39     public ensureRemain(remainSize:number){
  40         if(this.pos+remainSize>this.dv!.buffer.byteLength){
  41             let newSize=this.pos+remainSize;
  42             newSize+=newSize>>1;
  43             let buf=new ArrayBuffer(newSize);
  44             new Uint8Array(buf).set(new Uint8Array(this.dv!.buffer),0);
  45             this.dv=new DataView(buf);
  46         }
  47     }
  48     public putByte(val:number){
  49         this.dv!.setInt8(this.pos,val);
  50         this.pos++;
  51         return this;
  52     }
  53     public putInt(val:number){
  54         this.dv!.setInt32(this.pos,val,true);
  55         this.pos+=4;
  56         return this;
  57     }
  58
  59     public putLong(val:bigint){
  60         this.dv!.setBigInt64(this.pos,val,true);
  61         this.pos+=4;
  62         return this;
  63     }
  64     public putFloat(val:number){
  65         this.dv!.setFloat32(this.pos,val,true);
  66         this.pos+=4;
  67         return this;
  68     }
  69     public putDouble(val:number){
  70         this.dv!.setFloat64(this.pos,val,true);
  71         this.pos+=4;
  72         return this;
  73     }
  74     public putBytes(buf:ArrayBuffer){
  75         new Uint8Array(buf).set(new Uint8Array(buf),this.pos);
  76         this.pos+=buf.byteLength;
  77         return this;
  78     }
  79 }
  80
  81 export class Serilizer{
  82     protected buf?:ByteBuffer;
  83     public prepareUnseriling(buf:ArrayBuffer){
  84         this.buf=new ByteBuffer(buf);
  85         return this;
  86     }
  87     public prepareSerilizing(initBufSize:number){
  88         this.buf=new ByteBuffer(new ArrayBuffer(initBufSize));
  89         return this;
  90     }
  91     public putInt(val:number){
  92         this.buf!.ensureRemain(4);
  93         this.buf!.putInt(val);
  94         return this;
  95     }
  96     public putLong(val:bigint){
  97         this.buf!.ensureRemain(8);
  98         this.buf!.putLong(val);
  99         return this;
 100     }
 101     public putFloat(val:number){
 102         this.buf!.ensureRemain(4);
 103         this.buf!.putFloat(val);
 104         return this;
 105     }
 106     public putDouble(val:number){
 107         this.buf!.ensureRemain(8);
 108         this.buf!.putDouble(val);
 109         return this;
 110     }
 111
 112     public putBytes(b:ArrayBuffer){
 113         let len=b.byteLength;
 114         if(len>=0xff){
 115             this.buf!.ensureRemain(len+5);
 116             this.buf!.putByte(0xff).putInt(len);
 117         }else{
 118             this.buf!.ensureRemain(len+1);
 119             this.buf!.putByte(len);
 120         }
 121         this.buf!.putBytes(b);
 122         return this;
 123     }
 124     public putString(val:string){
 125         this.putBytes(new TextEncoder().encode(val));
 126         return this;
 127     }
 128     public build(){
 129         return this.buf!.dv!.buffer.slice(0,this.buf!.pos);
 130     }
 131     public getInt(){
 132         return this.buf!.getInt();
 133     }
 134     public getLong(){
 135         return this.buf!.getLong();
 136     }
 137     public getFloat(){
 138         return this.buf!.getFloat();
 139     }
 140     public getDouble(){
 141         return this.buf!.getDouble();
 142     }
 143     public getBytes(){
 144         let len=this.buf!.getByte();
 145         if(len==255){
 146             len=this.buf!.getInt();
 147         }
 148         return this.buf!.getBytes(len);
 149     }
 150     public getString(){
 151         return new TextDecoder().decode(this.getBytes());
 152     }
 153 }
 154
 155 export class RpcExtendError extends Error {
 156     public remoteException?:RpcExtendClientObject
 157 }
 158
 159 export class RpcExtendClientObject {
 160     public constructor(public client: RpcExtendClient1, public value: number | undefined) {
 161     }
 162     public async tryPull() {
 163         return this.client.conn.pull(this.value!);
 164     }
 165     public async free() {
 166         if (this.value != undefined) {
 167             await this.client.freeSlot(this.value);
 168         }
 169     }
 170     public async asCallable():Promise<RpcExtendClientCallable>{
 171         return new RpcExtendClientCallable(this.client,this.value)
 172     }
 173 }
 174
 175 export class RpcExtendClientCallable extends RpcExtendClientObject {
 176     protected sign: string = '->';
 177     public constructor(client: RpcExtendClient1, value: number | undefined) {
 178         super(client, value)
 179     }
 180     /*
 181 function signature
 182 format: 'parameters type->return type'
 183 eg:
 184 a function defined in c:
 185 bool fn(uin32_t,uint64_t,float64_t,struct pxprpc_object *)
 186 defined in java:
 187 boolean fn(int,int,double,Object)
 188 ...
 189 it's pxprpc signature:
 190 iido->z
 191
 192 available type signature characters:
 193 i  int(32bit integer)
 194 l  long(64bit integer)
 195 f  float(32bit float)
 196 d  double(64bit float)
 197 o  object(32bit reference address)
 198 b  bytes(32bit address refer to a bytes buffer)
 199 '' return void(32bit 0)
 200
 201 z  boolean(pxprpc use 32bit to store boolean value)
 202 s  string(bytes will be decode to string)
 203 */
 204     public signature(sign: string) {
 205         this.sign = sign;
 206         return this;
 207     }
 208
 209     public async call(...args:any[]) {
 210         let sign2 = this.sign
 211         let argSign = sign2.substring(0, sign2.lastIndexOf('->'))
 212         let retType = sign2.substring(argSign.length + 2)
 213
 214         let freeBeforeReturn: number[] = []
 215         //TODO: fix slightly memory waste.
 216         let args2 = new DataView(new ArrayBuffer(args.length * 8));
 217         let writeAt = 0;
 218         try {
 219             let argsProcessDone = false;
 220             for (let i1 = 0; i1 < argSign.length; i1++) {
 221                 switch (argSign.charAt(i1)) {
 222                     case 'i':
 223                         args2.setInt32(writeAt, args[i1], true);
 224                         writeAt += 4;
 225                         break;
 226                     case 'l':
 227                         args2.setBigInt64(writeAt, args[i1], true);
 228                         writeAt += 8;
 229                         break;
 230                     case 'f':
 231                         args2.setFloat32(writeAt, args[i1], true);
 232                         writeAt += 4;
 233                         break;
 234                     case 'd':
 235                         args2.setFloat64(writeAt, args[i1], true);
 236                         writeAt += 8;
 237                         break;
 238                     case 'o':
 239                         args2.setInt32(writeAt, args[i1].value, true);
 240                         writeAt += 4;
 241                         break;
 242                     case 's':
 243                     case 'b':
 244                         let t2 = await this.client.allocSlot()
 245                         freeBeforeReturn.push(t2)
 246                         if (argSign.charAt(i1) == 's') {
 247                             await this.client.conn.push(t2, new TextEncoder().encode(args[i1]))
 248                         } else {
 249                             await this.client.conn.push(t2, args[i1])
 250                         }
 251                         args2.setInt32(writeAt, t2, true);
 252                         writeAt += 4;
 253                         break;
 254                     case 'z':
 255                         args2.setInt32(writeAt, args[i1] ? 1 : 0, true);
 256                         writeAt += 4;
 257                         break;
 258                 }
 259                 if (argsProcessDone) {
 260                     break;
 261                 }
 262             }
 263             if (retType!=='' && 'ilfdz'.indexOf(retType) >= 0) {
 264                 let result = new DataView(await this.client.conn.call(
 265                     0, this.value!, args2.buffer.slice(0, writeAt),
 266                     'ld'.indexOf(retType) >= 0 ? 8 : 4));
 267                 let result2 = null;
 268                 switch (retType) {
 269                     case 'i':
 270                         result2 = result.getInt32(0, true);
 271                         break;
 272                     case 'l':
 273                         result2 = result.getBigInt64(0, true);
 274                         break;
 275                     case 'f':
 276                         result2 = result.getFloat32(0, true);
 277                         break;
 278                     case 'd':
 279                         result2 = result.getFloat64(0, true);
 280                         break;
 281                     case 'z':
 282                         result2 = result.getInt32(0, true) != 0;
 283                         break;
 284                 }
 285                 return result2
 286             }else{
 287                 let destAddr=await this.client.allocSlot()
 288                 let status = new DataView(await this.client.conn.call(
 289                     destAddr, this.value!, args2.buffer.slice(0, writeAt), 4)).getUint32(0,true);
 290                 let result=new RpcExtendClientObject(this.client,destAddr);
 291                 if(retType=='s'){
 292                     freeBeforeReturn.push(destAddr);
 293                     if(status==1){
 294                         await this.client.checkException(result);
 295                     }else{
 296                         let byteData=await result.tryPull()
 297                         if(byteData!=null){
 298                             let t2=new TextDecoder().decode(byteData);
 299                             return t2
 300                         }else{
 301                             return null;
 302                         }
 303                     }
 304                 }else if(retType=='b'){
 305                     freeBeforeReturn.push(destAddr);
 306                     if(status==1){
 307                         await this.client.checkException(result);
 308                     }else{
 309                         let byteData=await result.tryPull();
 310                         return byteData;
 311                     }
 312                 }else{
 313                     if(status==1){
 314                         await this.client.checkException(result);
 315                     }else{
 316                         return result
 317                     }
 318                 }
 319             }
 320         } finally {
 321             for (let t1 of freeBeforeReturn) {
 322                 await this.client.freeSlot(t1);
 323             }
 324         }
 325
 326     }
 327
 328 }
 329
 330
 331
 332 export class RpcExtendClient1 {
 333     private __usedSlots: { [index: number]: boolean | undefined } = {};
 334     private __nextSlots: number;
 335
 336     private __slotStart: number = 1;
 337     private __slotEnd: number = 64;
 338
 339     public constructor(public conn: Client) {
 340         this.__nextSlots = this.__slotStart;
 341     }
 342
 343     public setAvailableSlotsRange(start:number,end:number){
 344         this.__slotStart=start;
 345         this.__slotEnd=end;
 346         this.__nextSlots = this.__slotStart;
 347     }
 348
 349     public async init():Promise<this>{
 350         if(!this.conn.isRunning()){
 351             this.conn.run();
 352         }
 353         let info=await this.conn.getInfo();
 354         let refSlotsCap=info.split('\n').find(v=>v.startsWith('reference slots capacity:'))
 355         if(refSlotsCap!=undefined){
 356             this.setAvailableSlotsRange(1,Number(refSlotsCap.split(':')[1])-1);
 357         }
 358         return this;
 359     }
 360
 361     protected builtIn?:{checkException?:RpcExtendClientCallable}
 362     public async ensureBuiltIn(){
 363         if(this.builtIn==undefined){
 364             this.builtIn={}
 365             let t1=await this.getFunc('builtin.checkException');
 366             if(t1!=null){
 367                 t1.signature('o->s');
 368                 this.builtIn.checkException=t1;
 369             }
 370         }
 371     }
 372     public async checkException(obj:RpcExtendClientObject){
 373         await this.ensureBuiltIn();
 374         if(this.builtIn!.checkException!=null){
 375             let err=await this.builtIn!.checkException.call(obj) as string
 376             if(err!=''){
 377                 throw(new RpcExtendError(err));
 378             }
 379         }
 380     }
 381     public async allocSlot() {
 382         let reachEnd = false;
 383         while (this.__usedSlots[this.__nextSlots]) {
 384             this.__nextSlots += 1
 385             if (this.__nextSlots >= this.__slotEnd) {
 386                 if (reachEnd) {
 387                     throw new RpcExtendError('No slot available')
 388                 } else {
 389                     reachEnd = true
 390                     this.__nextSlots = this.__slotStart
 391                 }
 392             }
 393         }
 394
 395         let t1 = this.__nextSlots
 396         this.__nextSlots += 1
 397         if(this.__nextSlots>=this.__slotEnd){
 398             this.__nextSlots=this.__slotStart;
 399         }
 400         this.__usedSlots[t1] = true;
 401         return t1
 402     }
 403     public async freeSlot(index: number) {
 404         if (this.conn.isRunning()) {
 405             await this.conn.unlink(index)
 406             delete this.__usedSlots[index]
 407         }
 408
 409     }
 410     public async getFunc(name: string) {
 411         let t1 = await this.allocSlot()
 412         await this.conn.push(t1, new TextEncoder().encode(name))
 413         let t2 = await this.allocSlot()
 414         let resp=await this.conn.getFunc(t2, t1)
 415         await this.freeSlot(t1)
 416         if(resp==0){
 417             return null;
 418         }
 419         return new RpcExtendClientCallable(this, resp)
 420     }
 421     public async close(){
 422         for(let key in this.__usedSlots){
 423             if(this.__usedSlots[key])
 424                 this.freeSlot(Number(key));
 425         }
 426         await this.conn.close()
 427     }
 428
 429 }
 430
 431
 432 export class RpcExtendServerCallable implements PxpCallable{
 433     protected tParam:string = '';
 434     protected tResult:string='';
 435     protected paramBufLen=0;
 436     public constructor(public wrapped:(...args:any)=>Promise<any>){
 437     }
 438     //See RpcExtendClientCallable.signature
 439     public signature(sign: string) {
 440         let [tParam,tResult]=sign.split('->');
 441         this.tParam=tParam;
 442         this.tResult=tResult;
 443         this.paramBufLen=0;
 444         for(let i1=0;i1<tParam.length;i1++){
 445             if('ifobzs'.indexOf(tParam.charAt(i1))>=0){
 446                 this.paramBufLen+=4;
 447             }else if('ld'.indexOf(tParam.charAt(i1))>=0){
 448                 this.paramBufLen+=8;
 449             }
 450         }
 451         return this;
 452     }
 453     public async readParameter (req: PxpRequest){
 454         let buf=new DataView(await req.context.io1.read(this.paramBufLen));
 455         let tParam=this.tParam;
 456         let param=[];
 457         let offset=0;
 458         let obj:any=null;
 459         for(let i1=0;i1<tParam.length;i1++){
 460             switch(tParam[i1]){
 461                 case 'i':
 462                     param.push(buf.getInt32(offset,true));
 463                     offset+=4;
 464                     break;
 465                 case 'f':
 466                     param.push(buf.getFloat32(offset,true));
 467                     offset+=4;
 468                     break;
 469                 case 'o':
 470                 case 'b':
 471                     obj=req.context.refSlots[buf.getInt32(offset,true)]!.get();
 472                     param.push(obj);
 473                     offset+=4;
 474                     break;
 475                 case 's':
 476                     obj=req.context.refSlots[buf.getInt32(offset,true)]!.get();
 477                     if(obj instanceof ArrayBuffer){
 478                         param.push(new TextDecoder().decode(obj));
 479                     }else{
 480                         param.push(obj);
 481                     }
 482                     offset+=4;
 483                     break;
 484                 case 'l':
 485                     param.push(buf.getBigInt64(offset,true));
 486                     offset+=8;
 487                     break;
 488                 case 'd':
 489                     param.push(buf.getFloat64(offset,true));
 490                     offset+=8;
 491                     break;
 492                 default:
 493                     throw new Error('Unsupported value type.');
 494             }
 495         }
 496         req.parameter=param;
 497     }
 498     public async call(req: PxpRequest) : Promise<any>{
 499         try{
 500             return await this.wrapped.apply(this,req.parameter);
 501         }catch(e){
 502             return e;
 503         }
 504     }
 505     public async writeResult(req: PxpRequest){
 506         let buf=new DataView(new ArrayBuffer(8));
 507         let len=0;
 508         switch(this.tResult){
 509             case 'i':
 510                 buf.setInt32(0,req.result,true);len=4;
 511                 break;
 512             case 'f':
 513                 buf.setFloat32(0,req.result,true);len=4;
 514                 break;
 515             case 'o':
 516             case 'b':
 517             case 's':
 518             case '':
 519                 if(req.result instanceof Error){
 520                     buf.setInt32(0,1,true);
 521                 }else{
 522                     buf.setInt32(0,0,true);
 523                 }
 524                 len=4;
 525                 break;
 526             case 'l':
 527                 buf.setBigInt64(0,req.result,true);len=8;
 528                 break;
 529             case 'd':
 530                 buf.setFloat64(0,req.result,true);len=8;
 531                 break;
 532             default:
 533                 throw new Error('Unsupported value type.');
 534         }
 535         await req.context.io1.write(buf.buffer.slice(0,len));
 536     }
 537 }
 538 var builtinServerFuncMap:{[k:string]:RpcExtendServerCallable}={
 539     'builtin.checkException':new RpcExtendServerCallable(async(e:any)=>(e instanceof Error)?e.message:'').signature('o->s')
 540 }
 541 export class RpcExtendServer1{
 542     public constructor(public serv:Server){
 543         serv.funcMap=(name)=>this.findFunc(name)
 544     }
 545     public async serve(){
 546         await this.serv.serve()
 547     }
 548     public extFuncMap={} as {[k:string]:RpcExtendServerCallable};
 549     public findFunc(name:string){
 550         return this.extFuncMap[name]??(builtinServerFuncMap[name]);
 551     }
 552     public addFunc(name:string,fn:RpcExtendServerCallable){
 553         this.extFuncMap[name]=fn;
 554         return this;
 555     }
 556
 557 }