<commit_before>import {clone, pipeP, map} from 'ramda'
import {printRequest} from 'apollo-client/transport/networkInterface'

function applyWares (ctx, wares) {
  return new Promise(function (resolve, reject) {
    const queue = function (funcs, scope) {
      const next = function () {
        if (funcs.length > 0) {
          const f = funcs.shift()
          f.applyMiddleware.apply(scope, [ctx, next])
        }
        else {
          resolve(ctx)
        }
      }
      next()
    }
    queue(wares.slice())
  })
}

function executeQuery(sockets, context) {
  const {request, options} = context
  const {uri, channel} = options
  const Socket = options.Socket || require('./phoenix').Socket

  if (! uri) throw "Missing options.uri"
  if (! channel) throw "Missing options.channel"
  if (! channel.topic) throw "Missing options.channel.topic"

  if (options.logger === true) {
    options.logger = (kind, msg, data) =>
      console.log(`Apollo websocket ${kind}: ${msg}`, data)
  }

  if(! sockets[uri]) {
    sockets[uri] = {
      conn: new Socket(uri, options),
      channels: {}
    }
  }

  const socket = sockets[uri]
  let chan = socket.channels[channel.topic]

  if(! chan) {
    chan = socket.channels[channel.topic] = {
      conn: null,
      queue: []
    }
    socket.conn.onOpen(joinChannel)
  }

  function joinChannel() {
    if (! socket.conn.isConnected()) {
      return socket.conn.connect()
    }
    if (! chan.conn) {
      chan.conn = socket.conn.channel(channel.topic, channel.params || {})
    }
    if (!chan.conn.isJoined() && !chan.conn.isJoining()) {
      chan.conn.join().receive("ok", _ => {
        const queue = chan.queue
        chan.queue = []
        map(performQuery, queue)
      })
    }
  }


  function performQuery ({context, resolve, reject}) {
    const msg = context.options.channel.in_msg || "gql"
    const payload = printRequest(context.request)
    chan.conn.push(msg, payload)
      .receive("ok", response => {
        resolve(response)
      })
      .receive("error", reasons => {
        reject(reasons)
      })
      .receive("timeout", _ => {
        reject("timeout")
      })
  }

  return new Promise(function (resolve, reject) {
    if (chan.conn && chan.conn.isJoined()) {
      performQuery({context, resolve, reject})
    } else {
      chan.queue.push({context, resolve, reject})
      joinChannel()
    }
  })
}

const responseData = ({response}) => {
  return new Promise(function (resolve, reject) {
    if (response.error){
      reject(response)
    } else if (response.data) {
      resolve(response)
    } else {
      reject('No response')
    }
  })
}

export function createNetworkInterface(ifaceOpts) {
  const sockets = {}
  const middlewares = []
  const afterwares = []

  const use = map(item => middlewares.push(item))
  const useAfter = map(item => afterwares.push(item))

  const requestMiddleware = (request) => {
    const options = clone(ifaceOpts)<commit_msg>Added error handling for socket and channel connect.<commit_after>import {clone, pipeP, map} from 'ramda'
import {printRequest} from 'apollo-client/transport/networkInterface'

function applyWares (ctx, wares) {
  return new Promise(function (resolve, reject) {
    const queue = function (funcs, scope) {
      const next = function () {
        if (funcs.length > 0) {
          const f = funcs.shift()
          f.applyMiddleware.apply(scope, [ctx, next])
        }
        else {
          resolve(ctx)
        }
      }
      next()
    }
    queue(wares.slice())
  })
}

function executeQuery(sockets, context) {
  const {request, options} = context
  const {uri, channel} = options
  const Socket = options.Socket || require('./phoenix').Socket

  if (! uri) throw "Missing options.uri"
  if (! channel) throw "Missing options.channel"
  if (! channel.topic) throw "Missing options.channel.topic"

  if (options.logger === true) {
    options.logger = (kind, msg, data) =>
      console.log(`Apollo websocket ${kind}: ${msg}`, data)
  }

  if(! sockets[uri]) {
    sockets[uri] = {
      conn: new Socket(uri, options),
      channels: {}
    }
  }

  const socket = sockets[uri]
  let chan = socket.channels[channel.topic]

  function joinChannel(reject) {
    if (! socket.conn.isConnected()) {
      return socket.conn.connect()
    }
    if (! chan.conn) {
      chan.conn = socket.conn.channel(channel.topic, channel.params || {})
    }
    if (!chan.conn.isJoined() && !chan.conn.isJoining()) {
      chan.conn.join().receive("ok", _ => {
        const queue = chan.queue
        chan.queue = []
        map(performQuery, queue)
      }).receive('error', err => {
        chan.conn.leave();
        chan.conn = null;
        reject(new Error(err));
      })
    }
  }


  function performQuery ({context, resolve, reject}) {
    const msg = context.options.channel.in_msg || "gql"
    const payload = printRequest(context.request)
    chan.conn.push(msg, payload)
      .receive("ok", response => {
        resolve(response)
      })
      .receive("error", reasons => {
        reject(reasons)
      })
      .receive("timeout", _ => {
        reject("timeout")
      })
  }

  return new Promise(function (resolve, reject) {
    if(! chan) {
      chan = socket.channels[channel.topic] = {
        conn: null,
        queue: []
      }
      socket.conn.onOpen(joinChannel)
      socket.conn.onError(err => {
        reject(new Error(err))
      })
    }
    if (chan.conn && chan.conn.isJoined()) {
      performQuery({context, resolve, reject})
    } else {
      chan.queue.push({context, resolve, reject})
      joinChannel(reject)
    }
  })
}

const responseData = ({response}) => {
  return new Promise(function (resolve, reject) {
    if (response.error){
      reject(response)
    } else if (response.data) {
      resolve(response)
    } else {
      reject('No response')
    }
  })
}

export function createNetworkInterface(ifaceOpts) {
  const sockets = {}
  const middlewares = []
  const afterwares = []

  const use = map(item => middlewares.push(item))
  const useAfter = map(item => afterwares.push(item))

  const requestMiddleware = (request) => {
    const options = clone(ifaceOpts)
{'fraction_non_alphanumeric': 0.1475489403009222, 'fraction_numerical': 0.00032357223750202233, 'mean_word_length': 2.5900116144018583, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 25, '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': '1143307', 'n_tokens_mistral': 1949, 'n_tokens_neox': 1899, 'n_words': 424}