grpc1.ts 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123
  1. import { Subject, take } from 'rxjs';
  2. import { Message, MessageLog, ConnectionRequest } from '../interfaces/general.interface';
  3. import { GrpcServiceMethod } from '../services/grpc.service.method';
  4. import { readFileSync } from 'fs';
  5. import { ServerClientManager } from '../services/server-client.service';
  6. // Subject for bidirectional communication
  7. const connectionService: ServerClientManager = new ServerClientManager(new GrpcServiceMethod())
  8. const messagesJSON: any = readFileSync('payload.json')
  9. let parsedMessages: any[] = JSON.parse(messagesJSON) // load the fake messages generated for this trial
  10. let targetserver: string = 'localhost:3001'
  11. let targetserver2: string = 'localhost:3002'
  12. let hostServer: string = 'localhost:3000'
  13. let array: any[] = [] // Used for testing
  14. let connectionRequest: ConnectionRequest = {
  15. server: {
  16. serverUrl: hostServer,
  17. connectionType: 'GRPC',
  18. messageToBePublishedfromApplication: new Subject<Message>()
  19. },
  20. client: {
  21. targetServer: targetserver,
  22. connectionType: 'GRPC',
  23. messageToBeReceivedFromRemote: new Subject<Message>()
  24. }
  25. }
  26. // Handler for the incoming Messages from the other side.
  27. connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
  28. next: request => {
  29. // Application logic comes here. This is where the asortment takes place, of decidiing whose messages it belongs of what it is
  30. if ((request.message as MessageLog).appData.msgPayload == 'Query') {
  31. generateFakeStreamResponse(request).subscribe({
  32. next: (responseMessage: Message) => {
  33. // console.log(`Processing request:${request.id}....`)
  34. connectionRequest.server.messageToBePublishedfromApplication.next(responseMessage)
  35. },
  36. error: error => console.error(error),
  37. complete: () => {
  38. console.log(`Stream request for ${request.id} is queued.`) // shpuld be indefinite
  39. }
  40. })
  41. } else {
  42. array.push(request)
  43. console.log(`Received messages from the other side: ${(request.message as MessageLog).appData.msgId}`)
  44. }
  45. },
  46. error: error => console.error(error),
  47. complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
  48. })
  49. connectionService.generateConnection(connectionRequest)
  50. /* Simple Test */
  51. let generateFakeMessagesToBePublished = stream().pipe(take(10))
  52. generateFakeMessagesToBePublished.subscribe({
  53. next: message => {
  54. let payload: Message = {
  55. id: hostServer,
  56. message: message
  57. }
  58. connectionRequest.server.messageToBePublishedfromApplication.next(payload)
  59. }
  60. })
  61. /* Complex Test: Expected out come, both must receive 14 message by the end. Havent try to disconnect.*/
  62. // setTimeout(() => {
  63. // let message = {
  64. // id: parsedMessages[10].appData.msgId,
  65. // message: parsedMessages[10] // Choose this number, because i purposely use the 11th message and change the msgPayload property to query to emulate a request
  66. // }
  67. // connectionRequest.server.messageToBePublishedfromApplication.next(message)
  68. // }, 3000)
  69. // setTimeout(() => {
  70. // let message = {
  71. // id: parsedMessages[11].appData.msgId,
  72. // message: parsedMessages[11]// Choose this number, because i purposely use the 11th message and change the msgPayload property to query to emulate a request
  73. // }
  74. // connectionRequest.server.messageToBePublishedfromApplication.next(message)
  75. // }, 4000)
  76. // setTimeout(() => {
  77. // console.log(`All received data: ${array.length}`)
  78. // }, 10000)
  79. // setTimeout(() => {
  80. // console.log(`All received data: ${array.length}`)
  81. // }, 20000)
  82. // this is just to publish an array of fake data as a Subject
  83. function stream(): Subject<any> {
  84. let result: Subject<any> = new Subject()
  85. let messages: any[] = parsedMessages
  86. let count = 0
  87. const intervalId = setInterval(() => {
  88. result.next(messages[count]);
  89. count++;
  90. if (count >= 1000) {
  91. clearInterval(intervalId);
  92. result.complete();
  93. }
  94. }, 500)
  95. return result
  96. }
  97. function generateFakeStreamResponse(request: any): Subject<any> {
  98. let res: Subject<any> = new Subject()
  99. stream().pipe(take(7)).subscribe({
  100. next: element => {
  101. let message = {
  102. id: request.id, // Caller's
  103. message: element
  104. }
  105. res.next(message)
  106. },
  107. error: error => console.error(error),
  108. complete: () => console.log(`Stream response for ${request.id} has been prepared.`)
  109. })
  110. return res
  111. }