import _ from 'highland'
import duplex from './object-duplex'
import merge from 'mout/object/merge'

export default (deps, opts) => {

  let connect = () => {
    let connection = deps.net.connect(opts.port, opts.host)
    connection.setEncoding('utf8')
    return connection
  }

  let connection = null

  let api = {
    player: (channel, opts) => {

      opts = merge({
        id: deps.uuid().replace(/\-/g,''),
        fromStart: true
      }, opts)

      connection = connect()

      let output = _(connection)
        .fork()
        .filter(x => x.match(/^msg/))
        .map(x => x.replace('msg ',''))
        .map(JSON.parse)

      _(connection)
        .fork()
        .invoke('match', [/^error\s(\S+)\s(.+)/])
        .compact()
        .map(x => {
          let message = x[2]
          let err = new Error(message)
          err.code = x[1]
          return err
        })
        .each(x => output.emit('error', x))

      connection.write(
        'consume ' + channel + ' ' +
        opts.id + ' ' +
        (opts.fromStart ? 'smallest' : 'largest') +
        '\n')

      output.ack = () => connection.write('ack\n')
      return output

    },
    appender: (topic) => {
      connection = connect()
      let input = _()
      input
        .map(x =>
          'send ' +
          topic + ' ' +
          deps.uuid().replace(/\-/g, '') + ' ' +
          JSON.stringify(x) + '\n'
        ).pipe(connection)

      _(connection)
        .invoke('match', [/^error\s(\S+)\s(.+)/])
        .compact()
        .map(x => {
          let message = x[2]
          let err = new Error(message)
          err.code = x[1]
          return err
        })
        .each(x => input.emit('error', x))

      return input
    },


  }

  return api

}

@@ -299,16 +299,251 @@
 = null%0A%0A
+  let stringsToErrors = () =%3E _.pipeline(%0A    _.invoke('match', %5B/%5Eerror%5Cs(%5CS+)%5Cs(.+)/%5D),%0A    _.compact(),%0A    _.map(x =%3E %7B%0A      let message = x%5B2%5D%0A      let err = new Error(message)%0A      err.code = x%5B1%5D%0A      return err%0A    %7D)%0A  )%0A%0A
   let ap
@@ -548,16 +548,16 @@
 api = %7B%0A
-
     play
@@ -918,212 +918,33 @@
    .
-invoke('match', %5B/%5Eerror%5Cs(%5CS+)%5Cs(.+)/%5D)%0A        .compact()%0A        .map(x =%3E %7B%0A          let message = x%5B2%5D%0A          let err = new Error(message)%0A          err.code = x%5B1%5D%0A          return err%0A        %7D
+through(stringsToErrors()
 )%0A  
@@ -1514,212 +1514,33 @@
    .
-invoke('match', %5B/%5Eerror%5Cs(%5CS+)%5Cs(.+)/%5D)%0A        .compact()%0A        .map(x =%3E %7B%0A          let message = x%5B2%5D%0A          let err = new Error(message)%0A          err.code = x%5B1%5D%0A          return err%0A        %7D
+through(stringsToErrors()
 )%0A  

remove duplication