# src/Connector.js

    import BinaryEncoder from './Util/Binary/Encoder.js'
    import BinaryDecoder from './Util/Binary/Decoder.js'

    import { sendSyncStep1, readSyncStep1 } from './MessageHandler/syncStep1.js'
    import { readSyncStep2 } from './MessageHandler/syncStep2.js'
    import { integrateRemoteStructs } from './MessageHandler/integrateRemoteStructs.js'

    import debug from 'debug'

    // TODO: rename Connector

    export default class AbstractConnector {
      constructor (y, opts) {
        this.y = y
        this.opts = opts
        if (opts.role == null || opts.role === 'master') {
          this.role = 'master'
        } else if (opts.role === 'slave') {
          this.role = 'slave'
        } else {
          throw new Error("Role must be either 'master' or 'slave'!")
        }
        this.log = debug('y:connector')
        this.logMessage = debug('y:connector-message')
        this._forwardAppliedStructs = opts.forwardAppliedOperations || false // TODO: rename
        this.role = opts.role
        this.connections = new Map()
        this.isSynced = false
        this.userEventListeners = []
        this.whenSyncedListeners = []
        this.currentSyncTarget = null
        this.debug = opts.debug === true
        this.broadcastBuffer = new BinaryEncoder()
        this.broadcastBufferSize = 0
        this.protocolVersion = 11
        this.authInfo = opts.auth || null
        this.checkAuth = opts.checkAuth || function () { return Promise.resolve('write') } // default is everyone has write access
        if (opts.maxBufferLength == null) {
          this.maxBufferLength = -1
        } else {
          this.maxBufferLength = opts.maxBufferLength
        }
      }

      reconnect () {
        this.log('reconnecting..')
      }

      disconnect () {
        this.log('discronnecting..')
        this.connections = new Map()
        this.isSynced = false
        this.currentSyncTarget = null
        this.whenSyncedListeners = []
        return Promise.resolve()
      }

      onUserEvent (f) {
        this.userEventListeners.push(f)
      }

      removeUserEventListener (f) {
        this.userEventListeners = this.userEventListeners.filter(g => f !== g)
      }

      userLeft (user) {
        if (this.connections.has(user)) {
          this.log('%s: User left %s', this.y.userID, user)
          this.connections.delete(user)
          // check if isSynced event can be sent now
          this._setSyncedWith(null)
          for (var f of this.userEventListeners) {
            f({
              action: 'userLeft',
              user: user
            })
          }
        }
      }

      userJoined (user, role, auth) {
        if (role == null) {
          throw new Error('You must specify the role of the joined user!')
        }
        if (this.connections.has(user)) {
          throw new Error('This user already joined!')
        }
        this.log('%s: User joined %s', this.y.userID, user)
        this.connections.set(user, {
          uid: user,
          isSynced: false,
          role: role,
          processAfterAuth: [],
          processAfterSync: [],
          auth: auth || null,
          receivedSyncStep2: false
        })
        let defer = {}
        defer.promise = new Promise(function (resolve) { defer.resolve = resolve })
        this.connections.get(user).syncStep2 = defer
        for (var f of this.userEventListeners) {
          f({
            action: 'userJoined',
            user: user,
            role: role
          })
        }
        this._syncWithUser(user)
      }

      // Execute a function _when_ we are connected.
      // If not connected, wait until connected
      whenSynced (f) {
        if (this.isSynced) {
          f()
        } else {
          this.whenSyncedListeners.push(f)
        }
      }

      _syncWithUser (userID) {
        if (this.role === 'slave') {
          return // "The current sync has not finished or this is controlled by a master!"
        }
        sendSyncStep1(this, userID)
      }

      _fireIsSyncedListeners () {
        if (!this.isSynced) {
          this.isSynced = true
          // It is safer to remove this!
          // call whensynced listeners
          for (var f of this.whenSyncedListeners) {
            f()
          }
          this.whenSyncedListeners = []
          this.y._setContentReady()
          this.y.emit('synced')
        }
      }

      send (uid, buffer) {
        const y = this.y
        if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) {
          throw new Error('Expected Message to be an ArrayBuffer or Uint8Array - don\'t use this method to send custom messages')
        }
        this.log('User%s to User%s: Send \'%y\'', y.userID, uid, buffer)
        this.logMessage('User%s to User%s: Send %Y', y.userID, uid, [y, buffer])
      }

      broadcast (buffer) {
        const y = this.y
        if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) {
          throw new Error('Expected Message to be an ArrayBuffer or Uint8Array - don\'t use this method to send custom messages')
        }
        this.log('User%s: Broadcast \'%y\'', y.userID, buffer)
        this.logMessage('User%s: Broadcast: %Y', y.userID, [y, buffer])
      }

      /*
        Buffer operations, and broadcast them when ready.
      */
      broadcastStruct (struct) {
        const firstContent = this.broadcastBuffer.length === 0
        if (firstContent) {
          this.broadcastBuffer.writeVarString(this.y.room)
          this.broadcastBuffer.writeVarString('update')
          this.broadcastBufferSize = 0
          this.broadcastBufferSizePos = this.broadcastBuffer.pos
          this.broadcastBuffer.writeUint32(0)
        }
        this.broadcastBufferSize++
        struct._toBinary(this.broadcastBuffer)
        if (this.maxBufferLength > 0 && this.broadcastBuffer.length > this.maxBufferLength) {
          // it is necessary to send the buffer now
          // cache the buffer and check if server is responsive
          const buffer = this.broadcastBuffer
          buffer.setUint32(this.broadcastBufferSizePos, this.broadcastBufferSize)
          this.broadcastBuffer = new BinaryEncoder()
          this.whenRemoteResponsive().then(() => {
            this.broadcast(buffer.createBuffer())
          })
        } else if (firstContent) {
          // send the buffer when all transactions are finished
          // (or buffer exceeds maxBufferLength)
          setTimeout(() => {
            if (this.broadcastBuffer.length > 0) {
              const buffer = this.broadcastBuffer
              buffer.setUint32(this.broadcastBufferSizePos, this.broadcastBufferSize)
              this.broadcast(buffer.createBuffer())
              this.broadcastBuffer = new BinaryEncoder()
            }
          }, 0)
        }
      }

      /*
       * Somehow check the responsiveness of the remote clients/server
       * Default behavior:
       *   Wait 100ms before broadcasting the next batch of operations
       *
       * Only used when maxBufferLength is set
       *
       */
      whenRemoteResponsive () {
        return new Promise(function (resolve) {
          setTimeout(resolve, 100)
        })
      }

      /*
        You received a raw message, and you know that it is intended for Yjs. Then call this function.
      */
      receiveMessage (sender, buffer, skipAuth) {
        const y = this.y
        const userID = y.userID
        skipAuth = skipAuth || false
        if (!(buffer instanceof ArrayBuffer || buffer instanceof Uint8Array)) {
          return Promise.reject(new Error('Expected Message to be an ArrayBuffer or Uint8Array!'))
        }
        if (sender === userID) {
          return Promise.resolve()
        }
        let decoder = new BinaryDecoder(buffer)
        let encoder = new BinaryEncoder()
        let roomname = decoder.readVarString() // read room name
        encoder.writeVarString(roomname)
        let messageType = decoder.readVarString()
        let senderConn = this.connections.get(sender)
        this.log('User%s from User%s: Receive \'%s\'', userID, sender, messageType)
        this.logMessage('User%s from User%s: Receive %Y', userID, sender, [y, buffer])
        if (senderConn == null && !skipAuth) {
          throw new Error('Received message from unknown peer!')
        }
        if (messageType === 'sync step 1' || messageType === 'sync step 2') {
          let auth = decoder.readVarUint()
          if (senderConn.auth == null) {
            senderConn.processAfterAuth.push([messageType, senderConn, decoder, encoder, sender])
            // check auth
            return this.checkAuth(auth, y, sender).then(authPermissions => {
              if (senderConn.auth == null) {
                senderConn.auth = authPermissions
                y.emit('userAuthenticated', {
                  user: senderConn.uid,
                  auth: authPermissions
                })
              }
              let messages = senderConn.processAfterAuth
              senderConn.processAfterAuth = []

              messages.forEach(m =>
                this.computeMessage(m[0], m[1], m[2], m[3], m[4])
              )
            })
          }
        }
        if ((skipAuth || senderConn.auth != null) && (messageType !== 'update' || senderConn.isSynced)) {
          this.computeMessage(messageType, senderConn, decoder, encoder, sender, skipAuth)
        } else {
          senderConn.processAfterSync.push([messageType, senderConn, decoder, encoder, sender, false])
        }
      }

      computeMessage (messageType, senderConn, decoder, encoder, sender, skipAuth) {
        if (messageType === 'sync step 1' && (senderConn.auth === 'write' || senderConn.auth === 'read')) {
          // cannot wait for sync step 1 to finish, because we may wait for sync step 2 in sync step 1 (->lock)
          readSyncStep1(decoder, encoder, this.y, senderConn, sender)
        } else {
          const y = this.y
          y.transact(function () {
            if (messageType === 'sync step 2' && senderConn.auth === 'write') {
              readSyncStep2(decoder, encoder, y, senderConn, sender)
            } else if (messageType === 'update' && (skipAuth || senderConn.auth === 'write')) {
              integrateRemoteStructs(y, decoder)
            } else {
              throw new Error('Unable to receive message')
            }
          }, true)
        }
      }

      _setSyncedWith (user) {
        if (user != null) {
          const userConn = this.connections.get(user)
          userConn.isSynced = true
          const messages = userConn.processAfterSync
          userConn.processAfterSync = []
          messages.forEach(m => {
            this.computeMessage(m[0], m[1], m[2], m[3], m[4])
          })
        }
        const conns = Array.from(this.connections.values())
        if (conns.length > 0 && conns.every(u => u.isSynced)) {
          this._fireIsSyncedListeners()
        }
      }
    }