Packages
k8s
2.0.0-rc.6
2.8.0
2.7.0
2.6.2
2.6.1
2.6.0
2.5.0
2.4.2
2.4.1
2.4.0
2.3.0
2.2.0
2.1.1
2.1.0
2.0.3
2.0.2
2.0.1
2.0.0
2.0.0-rc.6
2.0.0-rc.5
2.0.0-rc.4
2.0.0-rc.3
2.0.0-rc.2
2.0.0-rc.1
2.0.0-rc.0
1.2.1
1.2.0
1.1.10
1.1.9
1.1.8
1.1.7
1.1.6
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.0
1.0.0-rc1
0.5.2
0.5.1
0.5.0
0.5.0-rc.2
0.5.0-rc.1
0.4.0
0.3.2
0.3.1
0.3.0
0.2.13
0.2.12
0.2.11
0.2.10
0.2.9
0.2.8
0.2.7
retired
0.2.6
0.2.5
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.3
Kubernetes API Client for Elixir
Current section
Files
Jump to
Current section
Files
lib/k8s/client/http_stream.ex
defmodule K8s.Client.HTTPStream do
@moduledoc """
Helper functions for HTTP stream processing.
"""
require Logger
alias K8s.Client.Provider
@type to_lines_t :: %{remainder: binary()}
@spec transform_to_lines(Enumerable.t(Provider.http_chunk_t())) ::
Enumerable.t(Provider.http_chunk_t() | {:line, binary()})
def transform_to_lines(stream) do
stream
|> Stream.transform(%{remainder: ""}, fn
{:data, chunk}, state ->
chunks_to_lines(chunk, state)
:done, %{remainder: ""} = state ->
{[:done], state}
:done, state ->
{[{:line, state.remainder}, :done], state}
other, state ->
{[other], state}
end)
end
@spec chunks_to_lines(binary(), to_lines_t()) :: {[{:line, binary()}], to_lines_t()}
defp chunks_to_lines(chunk, state) do
{remainder, whole_lines} =
(state.remainder <> chunk)
|> String.split("\n")
|> List.pop_at(-1)
resp = Enum.map(whole_lines, &{:line, &1})
{resp, Map.put(state, :remainder, remainder)}
end
@spec decode_json_objects(Enumerable.t(Provider.http_chunk_t())) ::
Enumerable.t(Provider.http_chunk_t() | {:object, binary()})
def decode_json_objects(stream) do
stream
|> transform_to_lines()
|> Stream.flat_map(fn
{:line, line} ->
case Jason.decode(line) do
{:error, error} ->
Logger.error(
"Could not decode JSON - chunk seems to be malformed",
library: :k8s,
error: error
)
[]
{:ok, object} ->
[{:object, object}]
end
other ->
[other]
end)
end
end