WebSocketConnection.js 5.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. const cleanPeriod = 60*1000;//1 минута
  2. class WebSocketConnection {
  3. //messageLifeTime в минутах (cleanPeriod)
  4. constructor(messageLifeTime = 5) {
  5. this.ws = null;
  6. this.timer = null;
  7. this.listeners = [];
  8. this.messageQueue = [];
  9. this.messageLifeTime = messageLifeTime;
  10. this.requestId = 0;
  11. }
  12. addListener(listener) {
  13. if (this.listeners.indexOf(listener) < 0)
  14. this.listeners.push(Object.assign({regTime: Date.now()}, listener));
  15. }
  16. //рассылаем сообщение и удаляем те обработчики, которые его получили
  17. emit(mes, isError) {
  18. const len = this.listeners.length;
  19. if (len > 0) {
  20. let newListeners = [];
  21. for (const listener of this.listeners) {
  22. let emitted = false;
  23. if (isError) {
  24. if (listener.onError)
  25. listener.onError(mes);
  26. emitted = true;
  27. } else {
  28. if (listener.onMessage) {
  29. if (listener.requestId) {
  30. if (listener.requestId === mes.requestId) {
  31. listener.onMessage(mes);
  32. emitted = true;
  33. }
  34. } else {
  35. listener.onMessage(mes);
  36. emitted = true;
  37. }
  38. } else {
  39. emitted = true;
  40. }
  41. }
  42. if (!emitted)
  43. newListeners.push(listener);
  44. }
  45. this.listeners = newListeners;
  46. }
  47. return this.listeners.length != len;
  48. }
  49. open(url) {
  50. return new Promise((resolve, reject) => {
  51. if (this.ws && this.ws.readyState == WebSocket.OPEN) {
  52. resolve(this.ws);
  53. } else {
  54. url = url || `ws://${window.location.host}/ws`;
  55. this.ws = new WebSocket(url);
  56. if (this.timer) {
  57. clearTimeout(this.timer);
  58. }
  59. this.timer = setTimeout(() => { this.periodicClean(); }, cleanPeriod);
  60. let resolved = false;
  61. this.ws.onopen = (e) => {
  62. resolved = true;
  63. resolve(e);
  64. };
  65. this.ws.onmessage = (e) => {
  66. try {
  67. const mes = JSON.parse(e.data);
  68. this.messageQueue.push({regTime: Date.now(), mes});
  69. let newMessageQueue = [];
  70. for (const message of this.messageQueue) {
  71. if (!this.emit(message.mes)) {
  72. newMessageQueue.push(message);
  73. }
  74. }
  75. this.messageQueue = newMessageQueue;
  76. } catch (e) {
  77. this.emit(e.message, true);
  78. }
  79. };
  80. this.ws.onerror = (e) => {
  81. this.emit(e.message, true);
  82. if (!resolved)
  83. reject(e);
  84. };
  85. }
  86. });
  87. }
  88. //timeout в минутах (cleanPeriod)
  89. message(requestId, timeout = 2) {
  90. return new Promise((resolve, reject) => {
  91. this.addListener({
  92. requestId,
  93. timeout,
  94. onMessage: (mes) => {
  95. resolve(mes);
  96. },
  97. onError: (e) => {
  98. reject(e);
  99. }
  100. });
  101. });
  102. }
  103. send(req) {
  104. if (this.ws && this.ws.readyState == WebSocket.OPEN) {
  105. const requestId = ++this.requestId;
  106. this.ws.send(JSON.stringify(Object.assign({requestId}, req)));
  107. return requestId;
  108. } else {
  109. throw new Error('WebSocket connection is not ready');
  110. }
  111. }
  112. close() {
  113. if (this.ws && this.ws.readyState == WebSocket.OPEN) {
  114. this.ws.close();
  115. }
  116. }
  117. periodicClean() {
  118. try {
  119. this.timer = null;
  120. const now = Date.now();
  121. //чистка listeners
  122. let newListeners = [];
  123. for (const listener of this.listeners) {
  124. if (now - listener.regTime < listener.timeout*cleanPeriod - 50) {
  125. newListeners.push(listener);
  126. } else {
  127. if (listener.onError)
  128. listener.onError('Время ожидания ответа истекло');
  129. }
  130. }
  131. this.listeners = newListeners;
  132. //чистка messageQueue
  133. let newMessageQueue = [];
  134. for (const message of this.messageQueue) {
  135. if (now - message.regTime < this.messageLifeTime*cleanPeriod - 50) {
  136. newMessageQueue.push(message);
  137. }
  138. }
  139. this.messageQueue = newMessageQueue;
  140. } finally {
  141. if (this.ws.readyState == WebSocket.OPEN) {
  142. this.timer = setTimeout(() => { this.periodicClean(); }, cleanPeriod);
  143. }
  144. }
  145. }
  146. }
  147. export default WebSocketConnection;