server-client.service.ts 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151
  1. import { BehaviorSubject, Subject } from 'rxjs';
  2. import { ClientRequest, ConnectionAttribute, ConnectionRequest, ConnectionState, Message, OutGoingInfo, ServerRequest, State } from '../interfaces/general.interface';
  3. import { GrpcServiceMethod } from './grpc.service.method';
  4. import { BufferService } from './buffer.service';
  5. import * as dotenv from 'dotenv'
  6. import * as _ from 'lodash'
  7. dotenv.config()
  8. export class ServerClientManager {
  9. private connectionAttributes: ConnectionAttribute[] = []
  10. private grpcService: GrpcServiceMethod = new GrpcServiceMethod()
  11. private defaultServerAttribute: ServerRequest = {
  12. name: 'Default - Server',
  13. serverUrl: "localhost:3000",
  14. connectionType: 'GRPC',
  15. messageToBePublishedFromApplication: new Subject<Message>()
  16. }
  17. private defaultClientAttribute: ClientRequest = {
  18. name: 'Default - Client',
  19. targetServer: "localhost:3001",
  20. connectionType: 'GRPC',
  21. messageToBeReceivedFromRemote: new Subject<Message>()
  22. }
  23. constructor() {
  24. // logic here
  25. }
  26. public async generateConnection(request: ConnectionRequest): Promise<any> {
  27. return new Promise(async (resolve, reject) => {
  28. let initialReport: ConnectionState
  29. let reportSubject: BehaviorSubject<ConnectionState>
  30. let retransmission: BufferService
  31. let errorString: string
  32. // let originalRequest = JSON.parse(JSON.stringify(request))
  33. let originalRequest = _.cloneDeep(request)
  34. let database: string
  35. let response: any = { message: `Fail to complete connection generation` }
  36. let statusChain: State = 1
  37. let connectionAttribute: ConnectionAttribute
  38. if (statusChain == 1) {
  39. if (!request.server) {
  40. request.server = this.defaultServerAttribute
  41. }
  42. if (!request.client) {
  43. request.client = this.defaultClientAttribute
  44. }
  45. if (request.database) {
  46. database = request.database
  47. } else {
  48. database = request.server.name + request.client.name
  49. }
  50. /* Inject retransmission here */
  51. initialReport = { status: 'BUFFER' }
  52. reportSubject = new BehaviorSubject(initialReport)
  53. retransmission = new BufferService(request.server.messageToBePublishedFromApplication, reportSubject, database)
  54. }
  55. if (statusChain == 1) {
  56. // Connection Type checking
  57. if (request.server!.connectionType != request.client!.connectionType) {
  58. console.log(`Connection Type DOES NOT MATCH!`)
  59. statusChain = -1
  60. errorString ="Connection Type DOES NOT MATCH!"
  61. } else {
  62. statusChain = 1
  63. }
  64. }
  65. if (statusChain == 1) {
  66. connectionAttribute = {
  67. ConnectionID: {
  68. local: request.server!.name + request.client!.name,
  69. remote: request.client!.name + request.server!.name
  70. },
  71. outGoing: {
  72. StreamID: request.server!.name,
  73. PublisherID: request.server!.name,
  74. SubscriberID: request.server!.name,
  75. MessageToBePublished: retransmission!.getMessages(),
  76. MessageToBeReceived: null
  77. },
  78. inComing: {
  79. StreamID: request.client!.name,
  80. PublisherID: request.client!.name,
  81. SubscriberID: request.client!.name,
  82. MessageToBePublished: null,
  83. MessageToBeReceived: request.client!.messageToBeReceivedFromRemote
  84. },
  85. connectionStatus: reportSubject!
  86. }
  87. statusChain = 1
  88. }
  89. if (statusChain == 1) {
  90. await this.checkConnectionAttribute(connectionAttribute!).then((res) => {
  91. if (res == true) {
  92. console.log(`Connection<${connectionAttribute.ConnectionID.local}> already exists `)
  93. }
  94. if (res == false) {
  95. this.connectionAttributes.push(connectionAttribute)
  96. console.log(`Connection ${connectionAttribute.ConnectionID.local} registered...`)
  97. }
  98. console.log(`There is now ${this.connectionAttributes.length} connection Attributes`)
  99. })
  100. statusChain = 1
  101. }
  102. if (statusChain == 1) {
  103. response = {
  104. message: "Channel Response",
  105. requestedTo: originalRequest,
  106. data: connectionAttribute!
  107. }
  108. resolve(response);
  109. } else if (statusChain == -1) {
  110. response = {
  111. message: "Channel Response Error",
  112. requestedTo: originalRequest,
  113. data: errorString! // put error string here
  114. }
  115. resolve(response);
  116. }
  117. if (statusChain == 1) {
  118. // This is default connection`
  119. if (!request.client!.connectionType) {
  120. request.client!.connectionType = 'GRPC'
  121. }
  122. // For each connection type:
  123. if (request.client!.connectionType == 'GRPC') {
  124. // this.grpcService.create(request, connectionAttribute, this.outGoingInfo)
  125. }
  126. }
  127. })
  128. }
  129. private async checkConnectionAttribute(connectionAttribute: ConnectionAttribute): Promise<boolean> {
  130. return new Promise((resolve) => {
  131. let result: boolean = this.connectionAttributes.some(connection =>
  132. connection.ConnectionID.local === connectionAttribute.ConnectionID.local
  133. );
  134. console.log(`Checking ${connectionAttribute.ConnectionID.local} and returns ${result}`);
  135. resolve(result);
  136. });
  137. }
  138. }