grpc1.ts 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355
  1. import { Subject, from, interval, 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. import mongoose from 'mongoose';
  7. // Connect to MongoDB
  8. mongoose.connect('mongodb://localhost:27017/grpc1')
  9. const Message = mongoose.model('Message', require('../models/message.schema'))
  10. // Subject for bidirectional communication
  11. const connectionService: ServerClientManager = new ServerClientManager()
  12. const messagesJSON: any = readFileSync('payload.json')
  13. let parsedMessages: any[] = JSON.parse(messagesJSON) // load the fake messages generated for this trial
  14. let targetserver: string = 'localhost:3001'
  15. let targetserver2: string = 'localhost:3002'
  16. let hostServer: string = 'localhost:3000'
  17. let array: any[] = [] // Used for testing
  18. let intervalToStreamOutGoingMessage: number = 1
  19. /* Simple Test: 1 to 1 */
  20. // let connectionRequest: ConnectionRequest = {
  21. // server: {
  22. // name: 'g1',
  23. // serverUrl: hostServer,
  24. // connectionType: 'GRPC',
  25. // messageToBePublishedFromApplication: new Subject<Message>()
  26. // },
  27. // client: {
  28. // name: 'g2',
  29. // targetServer: targetserver,
  30. // connectionType: 'GRPC',
  31. // messageToBeReceivedFromRemote: new Subject<Message>()
  32. // }
  33. // }
  34. // connectionService.generateConnection(connectionRequest)
  35. // let generateFakeMessagesToBePublished = stream().pipe(take(1000))
  36. // let generateFakeMessagesToBePublished = from(parsedMessages).pipe(take(5000))
  37. // generateFakeMessagesToBePublished.subscribe({
  38. // next: message => {
  39. // let payload: Message = {
  40. // id: hostServer,
  41. // message: message
  42. // }
  43. // connectionRequest.server.messageToBePublishedFromApplication.next(payload)
  44. // },
  45. // error: error => console.error(error),
  46. // complete: () => console.log(`Completed`)
  47. // })
  48. // stream().subscribe({
  49. // next: message => {
  50. // let payload = {
  51. // id: hostServer,
  52. // message: message
  53. // }
  54. // connectionRequest.server.messageToBePublishedFromApplication.next(payload)
  55. // }
  56. // })
  57. // connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
  58. // next: response => {
  59. // console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
  60. // // Message.create(response)
  61. // array.push(response)
  62. // },
  63. // error: error => console.error(error),
  64. // complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
  65. // })
  66. /* Complex Test: 1 to 1*/
  67. // let connectionRequest: ConnectionRequest = {
  68. // server: {
  69. // name: 'g1',
  70. // serverUrl: hostServer,
  71. // connectionType: 'GRPC',
  72. // messageToBePublishedfromApplication: new Subject<Message>()
  73. // },
  74. // client: {
  75. // name: 'g2',
  76. // targetServer: targetserver,
  77. // connectionType: 'GRPC',
  78. // messageToBeReceivedFromRemote: new Subject<Message>()
  79. // }
  80. // }
  81. // connectionService.generateConnection(connectionRequest)
  82. // setTimeout(() => {
  83. // let message = {
  84. // id: parsedMessages[10].appData.msgId,
  85. // message: parsedMessages[10] // Choose this number, because i purposely use the 11th message and change the msgPayload property to query to emulate a request
  86. // }
  87. // connectionRequest.server.messageToBePublishedfromApplication.next(message)
  88. // }, 3000)
  89. // setTimeout(() => {
  90. // let message = {
  91. // id: parsedMessages[11].appData.msgId,
  92. // message: parsedMessages[11] // Choose this number, because i purposely use the 12th message and change the msgPayload property to query to emulate a request
  93. // }
  94. // connectionRequest.server.messageToBePublishedfromApplication.next(message)
  95. // }, 4000)
  96. // Handler for the incoming Messages from the other side.
  97. // connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
  98. // next: request => {
  99. // // Application logic comes here. This is where the asortment takes place, of decidiing whose messages it belongs of what it is
  100. // if ((request.message as MessageLog).appData.msgPayload == 'Query') {
  101. // generateFakeStreamResponse(request).subscribe({
  102. // next: (responseMessage: Message) => {
  103. // console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
  104. // connectionRequest.server.messageToBePublishedfromApplication.next(responseMessage)
  105. // },
  106. // error: error => console.error(error),
  107. // complete: () => {
  108. // console.log(`Stream request for ${request.id} is queued.`) // shpuld be indefinite
  109. // }
  110. // })
  111. // } else {
  112. // array.push(request)
  113. // console.log(`Received message: ${(request.message as MessageLog).appData.msgId} from ${request.id}`)
  114. // }
  115. // },
  116. // error: error => console.error(error),
  117. // complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
  118. // })
  119. /* Simple Test: 1 to Many */
  120. let connectionRequest: ConnectionRequest = {
  121. server: {
  122. name: 'g1',
  123. serverUrl: hostServer,
  124. connectionType: 'GRPC',
  125. messageToBePublishedFromApplication: new Subject<Message>()
  126. },
  127. client: {
  128. name: 'g2',
  129. targetServer: targetserver,
  130. connectionType: 'GRPC',
  131. messageToBeReceivedFromRemote: new Subject<Message>()
  132. }
  133. }
  134. let connectionRequest2: ConnectionRequest = {
  135. server: {
  136. name: 'g1',
  137. serverUrl: hostServer,
  138. connectionType: 'GRPC',
  139. messageToBePublishedFromApplication: new Subject<Message>()
  140. },
  141. client: {
  142. name: 'g3',
  143. targetServer: targetserver2,
  144. connectionType: 'GRPC',
  145. messageToBeReceivedFromRemote: new Subject<Message>()
  146. }
  147. }
  148. connectionService.generateConnection(connectionRequest)
  149. connectionService.generateConnection(connectionRequest2)
  150. let generateFakeMessagesToBePublished = from(parsedMessages).pipe(take(50))
  151. generateFakeMessagesToBePublished.subscribe({
  152. next: message => {
  153. let payload: Message = {
  154. id: hostServer,
  155. message: message
  156. }
  157. connectionRequest.server.messageToBePublishedFromApplication.next(payload)
  158. connectionRequest2.server.messageToBePublishedFromApplication.next(payload)
  159. },
  160. error: error => console.error(error),
  161. complete: () => console.log(`Completed`)
  162. })
  163. // let generateFakeMessagesToBePublished = stream().pipe(take(10))
  164. // generateFakeMessagesToBePublished.subscribe({
  165. // next: message => {
  166. // let payload: Message = {
  167. // id: hostServer,
  168. // message: message
  169. // }
  170. // connectionRequest.server.messageToBePublishedFromApplication.next(payload)
  171. // connectionRequest2.server.messageToBePublishedFromApplication.next(payload)
  172. // }
  173. // })
  174. connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
  175. next: request => {
  176. console.log(`Received ${(request.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
  177. array.push(request)
  178. },
  179. error: error => console.error(error),
  180. complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
  181. })
  182. connectionRequest2.client.messageToBeReceivedFromRemote.subscribe({
  183. next: request => {
  184. console.log(`Received ${(request.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
  185. array.push(request)
  186. },
  187. error: error => console.error(error),
  188. complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
  189. })
  190. /* Complex Test: 1 to Many */
  191. // let connectionRequest: ConnectionRequest = {
  192. // server: {
  193. // name: 'g1',
  194. // serverUrl: hostServer,
  195. // connectionType: 'GRPC',
  196. // messageToBePublishedfromApplication: new Subject<Message>()
  197. // },
  198. // client: {
  199. // name: 'g2',
  200. // targetServer: targetserver,
  201. // connectionType: 'GRPC',
  202. // messageToBeReceivedFromRemote: new Subject<Message>()
  203. // }
  204. // }
  205. // let connectionRequest2: ConnectionRequest = {
  206. // server: {
  207. // name: 'g1',
  208. // serverUrl: hostServer,
  209. // connectionType: 'GRPC',
  210. // messageToBePublishedfromApplication: new Subject<Message>()
  211. // },
  212. // client: {
  213. // name: 'g3',
  214. // targetServer: targetserver2,
  215. // connectionType: 'GRPC',
  216. // messageToBeReceivedFromRemote: new Subject<Message>()
  217. // }
  218. // }
  219. // connectionService.generateConnection(connectionRequest)
  220. // connectionService.generateConnection(connectionRequest2)
  221. // setTimeout(() => {
  222. // let message = {
  223. // id: parsedMessages[10].appData.msgId,
  224. // message: parsedMessages[10] // Choose this number, because i purposely use the 11th message and change the msgPayload property to query to emulate a request
  225. // }
  226. // connectionRequest.server.messageToBePublishedfromApplication.next(message)
  227. // connectionRequest2.server.messageToBePublishedfromApplication.next(message)
  228. // }, 3000)
  229. // setTimeout(() => {
  230. // let message = {
  231. // id: parsedMessages[11].appData.msgId,
  232. // message: parsedMessages[11] // Choose this number, because i purposely use the 12th message and change the msgPayload property to query to emulate a request
  233. // }
  234. // connectionRequest.server.messageToBePublishedfromApplication.next(message)
  235. // connectionRequest2.server.messageToBePublishedfromApplication.next(message)
  236. // }, 4000)
  237. // connectionRequest.client.messageToBeReceivedFromRemote.subscribe({
  238. // next: request => {
  239. // // Application logic comes here. This is where the asortment takes place, of deciding whose messages it belongs of what it is
  240. // if ((request.message as MessageLog).appData.msgPayload == 'Query') {
  241. // generateFakeStreamResponse(request).subscribe({
  242. // next: (responseMessage: Message) => {
  243. // connectionRequest.server.messageToBePublishedfromApplication.next(responseMessage)
  244. // },
  245. // error: error => console.error(error),
  246. // complete: () => {
  247. // console.log(`Stream request for ${request.id} is queued.`) // should be indefinite
  248. // }
  249. // })
  250. // } else {
  251. // console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
  252. // array.push(request)
  253. // console.log(`Received message: ${(request.message as MessageLog).appData.msgId} from ${request.id}`)
  254. // }
  255. // },
  256. // error: error => console.error(error),
  257. // complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
  258. // })
  259. // connectionRequest2.client.messageToBeReceivedFromRemote.subscribe({
  260. // next: request => {
  261. // // Application logic comes here. This is where the asortment takes place, of deciding whose messages it belongs of what it is
  262. // if ((request.message as MessageLog).appData.msgPayload == 'Query') {
  263. // generateFakeStreamResponse(request).subscribe({
  264. // next: (responseMessage: Message) => {
  265. // connectionRequest2.server.messageToBePublishedfromApplication.next(responseMessage)
  266. // },
  267. // error: error => console.error(error),
  268. // complete: () => {
  269. // console.log(`Stream request for ${request.id} is queued.`) // should be indefinite
  270. // }
  271. // })
  272. // } else {
  273. // console.log(`Received ${(response.message as MessageLog).appData.msgId} from ${connectionRequest.client.targetServer}`)
  274. // array.push(request)
  275. // console.log(`Received message: ${(request.message as MessageLog).appData.msgId} from ${request.id}`)
  276. // }
  277. // },
  278. // error: error => console.error(error),
  279. // complete: () => console.log(`Response for incoming generated. But this will never stop, and should not either.`)
  280. // })
  281. // this is just to publish an array of fake data as a Subject
  282. function stream(): Subject<any> {
  283. let result: Subject<any> = new Subject()
  284. let messages: any[] = parsedMessages
  285. let count = 0
  286. const intervalId = setInterval(() => {
  287. result.next(messages[count]);
  288. count++;
  289. if (count >= 1000) {
  290. clearInterval(intervalId);
  291. result.complete();
  292. }
  293. }, intervalToStreamOutGoingMessage)
  294. return result
  295. }
  296. function generateFakeStreamResponse(request: any): Subject<any> {
  297. let res: Subject<any> = new Subject()
  298. stream().pipe(take(7)).subscribe({
  299. next: element => {
  300. let message = {
  301. id: request.id, // Caller's
  302. message: element
  303. }
  304. res.next(message)
  305. },
  306. error: error => console.error(error),
  307. complete: () => console.log(`Stream response for ${request.id} has been prepared.`)
  308. })
  309. return res
  310. }
  311. /* Checking the values by the end of the test */
  312. interval(5000).subscribe(() => {
  313. console.log(`All received data: ${array.length}`);
  314. });