<commit_before>exports.name = "creationix/coro-channel"
exports.version = "1.0.3"

-- Given a raw uv_stream_t userdara, return coro-friendly read/write functions.
-- Given a raw uv_stream_t userdara, return coro-friendly read/write functions.
function exports.wrapStream(socket)
  local paused = true
  local queue = {}
  local waiting
  local reading = true
  local writing = true

  local onRead

  local function read()
    if #queue > 0 then
      return unpack(table.remove(queue, 1))
    end
    if paused then
      paused = false
      assert(socket:read_start(onRead))
    end
    waiting = coroutine.running()
    return coroutine.yield()
  end

  local flushing = false
  local flushed = false
  local function checkShutdown()
    if socket:is_closing() then return end
    if not flushing and not writing then
      flushing = true
      local thread = coroutine.running()
      socket:shutdown(function (err)
        flushed = true
        coroutine.resume(thread, not err, err)
      end)
      coroutine.yield()
    end
    if flushed and not reading then
      socket:close()
    end
  end

  function onRead(err, chunk)
    assert(not err, err)
    local data = err and {nil, err} or {chunk}
    if waiting then
      local thread = waiting
      waiting = nil
      assert(coroutine.resume(thread, unpack(data)))
    else
      queue[#queue + 1] = data
      if not paused then
        paused = true
        assert(socket:read_stop())
      end
    end
    if not chunk then
      reading = false
      -- Close the whole socket if the writing side is also closed already.
      checkShutdown()
    end
  end

  local function write(chunk)<commit_msg>Import coro-channel fixes from rye<commit_after>exports.name = "creationix/coro-channel"
exports.version = "1.0.4"

-- Given a raw uv_stream_t userdara, return coro-friendly read/write functions.
-- Given a raw uv_stream_t userdara, return coro-friendly read/write functions.
function exports.wrapStream(socket)
  local paused = true
  local queue = {}
  local waiting
  local reading = true
  local writing = true

  local onRead

  local function read()
    if #queue > 0 then
      return unpack(table.remove(queue, 1))
    end
    if paused then
      paused = false
      assert(socket:read_start(onRead))
    end
    waiting = coroutine.running()
    return coroutine.yield()
  end

  local flushing = false
  local flushed = false
  local function checkShutdown()
    if socket:is_closing() then return end
    if not flushing and not writing then
      flushing = true
      local thread = coroutine.running()
      socket:shutdown(function (err)
        flushed = true
        coroutine.resume(thread, not err, err)
      end)
      assert(coroutine.yield())
    end
    if flushed and not reading then
      socket:close()
    end
  end

  function onRead(err, chunk)
    local data = err and {nil, err} or {chunk}
    if waiting then
      local thread = waiting
      waiting = nil
      assert(coroutine.resume(thread, unpack(data)))
    else
      queue[#queue + 1] = data
      if not paused then
        paused = true
        assert(socket:read_stop())
      end
    end
    if not chunk then
      reading = false
      -- Close the whole socket if the writing side is also closed already.
      checkShutdown()
    end
  end

  local function write(chunk)
{'fraction_non_alphanumeric': 0.07719928186714542, 'fraction_numerical': 0.003590664272890485, 'mean_word_length': 2.5754010695187164, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '1318028', 'n_tokens_mistral': 995, 'n_tokens_neox': 980, 'n_words': 338}