<commit_before>import { TOPIC, CONNECTION_ACTION, ParseResult, Message } from '../../constants'
import * as binaryMessageBuilder from '@deepstream/protobuf/dist/src/message-builder'
import * as binaryMessageParser from '@deepstream/protobuf/dist/src/message-parser'
import { WebSocketServerConfig } from '../websocket/connection-endpoint'
import { SocketConnectionEndpoint, StatefulSocketWrapper, DeepstreamServices, UnauthenticatedSocketWrapper, SocketWrapper, EVENT } from '../../../ds-types/src/index'
import * as WebSocket from 'ws'

/**
 * This class wraps around a websocket
 * and provides higher level methods that are integrated
 * with deepstream's message structure
 */
export class WSSocketWrapper implements UnauthenticatedSocketWrapper {

  public isRemote: false = false
  public isClosed: boolean = false
  public user: string | null = null
  public uuid: number = Math.random()
  public authCallback: Function | null = null
  public authAttempts: number = 0

  private bufferedWrites: Uint8Array[]
  private closeCallbacks: Set<Function> = new Set()

  public authData: object | null = null
  public clientData: object | null = null
  private bufferedWritesTotalByteSize: number

  constructor (
    private socket: WebSocket,
    private handshakeData: any,
    private services: DeepstreamServices,
    private config: WebSocketServerConfig,
    private connectionEndpoint: SocketConnectionEndpoint
   ) {
    this.bufferedWritesTotalByteSize = 0
    this.bufferedWrites = []
  }

  get isOpen () {
    return this.isClosed !== true
  }

  /**
   * Called by the connection endpoint to flush all buffered writes.
   * A buffered write is a write that is not a high priority, such as an ack
   * and can wait to be bundled into another message if necessary
   */
  public flush () {
    if (this.bufferedWritesTotalByteSize !== 0) {
      this.socket.send(this.bufferedWrites)
      this.bufferedWritesTotalByteSize = 0
      this.bufferedWrites = []
    }
  }

  /**
   * Sends a message based on the provided action and topic
   */
  public sendMessage (message: { topic: TOPIC, action: CONNECTION_ACTION } | Message, allowBuffering: boolean = true): void {
    this.sendBuiltMessage(binaryMessageBuilder.getMessage(message, false), allowBuffering)
  }

  /**
   * Sends a message based on the provided action and topic
   */
  public sendAckMessage (message: Message, allowBuffering: boolean = true): void {
    this.sendBuiltMessage(
        binaryMessageBuilder.getMessage(message, true),
        true<commit_msg>fix: unoptimial send loop for connection endpoint<commit_after>import { TOPIC, CONNECTION_ACTION, ParseResult, Message } from '../../constants'
import * as binaryMessageBuilder from '@deepstream/protobuf/dist/src/message-builder'
import * as binaryMessageParser from '@deepstream/protobuf/dist/src/message-parser'
import { WebSocketServerConfig } from '../websocket/connection-endpoint'
import { SocketConnectionEndpoint, StatefulSocketWrapper, DeepstreamServices, UnauthenticatedSocketWrapper, SocketWrapper, EVENT } from '../../../ds-types/src/index'
import * as WebSocket from 'ws'

/**
 * This class wraps around a websocket
 * and provides higher level methods that are integrated
 * with deepstream's message structure
 */
export class WSSocketWrapper implements UnauthenticatedSocketWrapper {

  public isRemote: false = false
  public isClosed: boolean = false
  public user: string | null = null
  public uuid: number = Math.random()
  public authCallback: Function | null = null
  public authAttempts: number = 0

  private bufferedWrites: Uint8Array[] = []
  private closeCallbacks: Set<Function> = new Set()

  public authData: object | null = null
  public clientData: object | null = null
  private bufferedWritesTotalByteSize: number = 0

  constructor (
    private socket: WebSocket,
    private handshakeData: any,
    private services: DeepstreamServices,
    private config: WebSocketServerConfig,
    private connectionEndpoint: SocketConnectionEndpoint
   ) {
  }

  get isOpen () {
    return this.isClosed !== true
  }

  /**
   * Called by the connection endpoint to flush all buffered writes.
   * A buffered write is a write that is not a high priority, such as an ack
   * and can wait to be bundled into another message if necessary
   */
  public flush () {
    if (this.bufferedWritesTotalByteSize !== 0) {
      this.bufferedWrites.forEach((bw) => this.socket.send(bw))
      this.bufferedWritesTotalByteSize = 0
      this.bufferedWrites = []
    }
  }

  /**
   * Sends a message based on the provided action and topic
   */
  public sendMessage (message: { topic: TOPIC, action: CONNECTION_ACTION } | Message, allowBuffering: boolean = true): void {
    this.sendBuiltMessage(binaryMessageBuilder.getMessage(message, false), allowBuffering)
  }

  /**
   * Sends a message based on the provided action and topic
   */
  public sendAckMessage (message: Message, allowBuffering: boolean = true): void {
    this.sendBuiltMessage(
        binaryMessageBuilder.getMessage(message, true),
        true
{'fraction_non_alphanumeric': 0.0880205249654628, 'fraction_numerical': 0.0019735543714229325, 'mean_word_length': 3.958904109589041, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4622153', 'n_tokens_mistral': 1396, 'n_tokens_neox': 1337, 'n_words': 448}