defmodule Hbasex do use Application alias Hbasex.Models.{TGet, TColumn, TColumnValue, TPut, TScan, TDelete} @default_host "localhost" @default_thrift_port 9090 @default_rest_port 8080 def start(_type, _args) do config = get_config() Hbasex.Pool.start_link(config) end defp get_config(config \\ nil) do config = if is_nil(config) do Application.get_all_env(:hbasex) |> Map.new() else config end config |> Map.put_new(:thrift_port, @default_thrift_port) |> Map.put_new(:rest_port, @default_rest_port) end def connect(config \\ %{}) do config |> get_config() |> Hbasex.Pool.connect() end def delete_table(config \\ nil, table_name) do config = get_config(config) Hbasex.RestClient.delete(config, table_name) end def create_table(config \\ nil, table_name, column_family) do config = get_config(config) Hbasex.RestClient.create(config, table_name, column_family) end def get(table_name, row, map \\ %{}) do tcolumns = for {family, qualifiers} <- map do for qualifier <- qualifiers do TColumn.new(family: family, qualifier: qualifier) end end |> List.flatten tget = TGet.new(attributes: :dict.new(), row: row, columns: tcolumns) Hbasex.Pool.exec(:get, [table_name, tget]) |> parse_row_result() |> elem(1) end def put(table_name, row, map) do Hbasex.Pool.exec(:put, [table_name, get_t_put(row, map)]) end def delete(table_name, row) when is_bitstring(row) do Hbasex.Pool.exec(:deleteSingle, [table_name, get_t_delete(row)]) end def delete(table_name, rows) when is_list(rows) do t_deletes = rows |> Enum.map(&get_t_delete/1) Hbasex.Pool.exec(:deleteMultiple, [table_name, t_deletes]) end def scan(table_name, nb_rows, options \\ []) do Hbasex.Pool.exec(:getScannerResults, [table_name, get_t_scan(options), nb_rows]) |> Enum.map(&parse_row_result/1) end defp get_t_put(row, map) do column_values = for {family, columns} <- map do for {qualifier, value} <- columns do TColumnValue.new(family: family, qualifier: qualifier, value: value) end end |> List.flatten TPut.new(attributes: :dict.new(), row: row, columnValues: column_values) end defp get_t_delete(row, args \\ []) do args = [{:row, row} | args] args = if is_nil(args[:attributes]) do [{:attributes, :dict.new()}| args] else args end TDelete.new(args) end defp get_t_scan(options) do options = expand_scan_options(options) options = if is_nil(options[:attributes]) do [{:attributes, :dict.new()}| options] else options end TScan.new(options) end defp parse_row_result(result) do { result.row, result.columnValues |> Enum.reduce(%{}, fn(x, acc) -> Map.update(acc, x.family, %{x.qualifier => x.value}, &(Map.put(&1, x.qualifier, x.value))) end) } end defp expand_scan_options(acc), do: expand_scan_options(acc, []) defp expand_scan_options([], acc), do: acc defp expand_scan_options([{:prefix, prefix}| rest], acc) do preprefix_size = byte_size(prefix) - 1 <> = prefix stop_row = preprefix <> <> acc = [{:startRow, prefix}, {:stopRow, stop_row} | acc] expand_scan_options(rest, acc) end defp expand_scan_options([kv | rest], acc) do expand_scan_options(rest, [kv | acc]) end end