///|
priv suberror PubSubCallbackError {
  PubSubCallbackError(Error)
}

///|
async fn Client::subscribe_resp2_channel_loop(
  self : Client,
  channel : String,
  callback : async (Message) -> Unit,
) -> Unit {
  @async.with_task_group(group => {
    group.spawn_loop(
      () => self.run_resp2_channel_subscription(channel, callback),
      retry=self.config.reconnect_strategy,
      fatal_error=is_pubsub_fatal_error,
    )
  }) catch {
    PubSubCallbackError(err) => raise err
    err => raise err
  }
}

///|
async fn Client::subscribe_resp2_pattern_loop(
  self : Client,
  pattern : String,
  callback : async (PatternMessage) -> Unit,
) -> Unit {
  @async.with_task_group(group => {
    group.spawn_loop(
      () => self.run_resp2_pattern_subscription(pattern, callback),
      retry=self.config.reconnect_strategy,
      fatal_error=is_pubsub_fatal_error,
    )
  }) catch {
    PubSubCallbackError(err) => raise err
    err => raise err
  }
}

///|
fn is_pubsub_fatal_error(err : Error) -> Bool {
  match err {
    ServerError(_)
    | HandshakeError(_)
    | UnexpectedResponse(_)
    | InvalidUtf8(_)
    | IntegerOverflow(_)
    | PubSubCallbackError(_) => true
    _ => false
  }
}

///|
async fn Client::handshake_resp2_pubsub(
  self : Client,
  r : @resp.Reader,
  w : @resp.Writer,
) -> Unit {
  if self.config.password is Some(password) {
    self.expect_startup_ok(
      w,
      r,
      ReadOnlyArray::from_array(self.legacy_auth_command(password)),
      version=RESP2,
    )
  }
  if self.config.database != 0 {
    self.expect_startup_ok(
      w,
      r,
      [b"SELECT", @encoding/utf8.encode(self.config.database.to_string())],
      version=RESP2,
    )
  }
  if self.config.name is Some(name) {
    self.expect_startup_ok(
      w,
      r,
      [b"CLIENT", b"SETNAME", @encoding/utf8.encode(name)],
      version=RESP2,
    )
  }
}

///|
async fn Client::run_resp2_channel_subscription(
  self : Client,
  channel : String,
  callback : async (Message) -> Unit,
) -> Unit {
  let conn = self.dial()
  defer conn.close()
  let reader = @resp.Reader(
    conn,
    buffer_size=self.config.read_buffer_size,
    max_depth=self.config.resp_max_depth,
  )
  let writer = @resp.Writer(conn, buffer_size=self.config.write_buffer_size)
  self.handshake_resp2_pubsub(reader, writer)
  writer.write_command(
    pubsub_command(kind=Channel, subscribe=true, name=channel),
  )
  for ;; {
    match reader.read_response(version=RESP2) {
      Reply({ desc: Error(err), .. }) => raise response_error(err)
      Reply(value) =>
        match value.desc {
          Array(values) =>
            match parse_pubsub_event(values) {
              None => ()
              Some(ChannelMessage(channel, payload)) =>
                callback({ channel, payload }) catch {
                  err if @async.is_cancellation_error(err) => raise err
                  err => raise PubSubCallbackError(err)
                }
              Some(PatternMessage(_, _, _)) => ()
            }
          _ => raise UnexpectedResponse("expected pubsub array")
        }
      Push(_) => raise @resp.ProtocolError
    }
  }
}

///|
async fn Client::run_resp2_pattern_subscription(
  self : Client,
  pattern : String,
  callback : async (PatternMessage) -> Unit,
) -> Unit {
  let conn = self.dial()
  defer conn.close()
  let reader = @resp.Reader(
    conn,
    buffer_size=self.config.read_buffer_size,
    max_depth=self.config.resp_max_depth,
  )
  let writer = @resp.Writer(conn, buffer_size=self.config.write_buffer_size)
  self.handshake_resp2_pubsub(reader, writer)
  writer.write_command(
    pubsub_command(kind=Pattern, subscribe=true, name=pattern),
  )
  for ;; {
    match reader.read_response(version=RESP2) {
      Reply({ desc: Error(err), .. }) => raise response_error(err)
      Reply(value) =>
        match value.desc {
          Array(values) =>
            match parse_pubsub_event(values) {
              None => ()
              Some(ChannelMessage(_, _)) => ()
              Some(PatternMessage(pattern, channel, payload)) =>
                callback({ pattern, channel, payload }) catch {
                  err if @async.is_cancellation_error(err) => raise err
                  err => raise PubSubCallbackError(err)
                }
            }
          _ => raise UnexpectedResponse("expected pubsub array")
        }
      Push(_) => raise @resp.ProtocolError
    }
  }
}