defmodule OpencadcVospaceClient do @moduledoc """ VOSpace client to upload a stream of data. Useful for transferring from one location to a VOSpace. """ @vos_namespace "http://www.ivoa.net/xml/VOSpace/v2.0" @doc """ Upload a file to a VOSpace `destination_uri`: VOS URI to put to. `cookie_credential`: Cookie value for auth. """ def upload(destination_uri, cookie_credential) do (OpencadcRegistryClient.lookup_service_url( "ivo://cadc.nrc.ca/vospace", "ivo://ivoa.net/std/VOSpace/v2.0#nodes", "vs:ParamHTTP" ) <> URI.parse(destination_uri).path) |> HTTPoison.put!( to_xml(destination_uri), [{"Content-Type", "text/xml"}, {"Cookie", "CADC_SSO=\"#{cookie_credential}\""}], [{:follow_redirect, true}, hackney: [:force_redirect, true]] ) |> write(destination_uri, cookie_credential) end defp write(%HTTPoison.Response{status_code: 200}, destination_uri, cookie_credential) do with transfer_service_url = OpencadcRegistryClient.lookup_service_url( "ivo://cadc.nrc.ca/vospace", "ivo://ivoa.net/std/VOSpace/v2.0#transfers", "vs:ParamHTTP" ), vospace_sync_url = OpencadcRegistryClient.lookup_service_url( "ivo://cadc.nrc.ca/vospace", "ivo://ivoa.net/std/VOSpace#sync-2.1", "vs:ParamHTTP" ), xml_body = to_xml(destination_uri, transfer_service_url) do HTTPoison.post!( vospace_sync_url, xml_body, [ {"Content-Type", "text/xml"}, {"Cookie", "CADC_SSO=\"#{cookie_credential}\""} ] ) |> upload_to(cookie_credential) end end defp write(%HTTPoison.Response{status_code: 401}, destination_uri, _) do raise ArgumentError, message: "Write permission denied on " <> destination_uri end defp write(%HTTPoison.Response{status_code: 400, body: body}, _, _) do raise ArgumentError, message: body end defp write(%HTTPoison.Response{status_code: 404}, destination_uri, cookie_credential) do with uri = URI.parse(destination_uri), parent_path = uri.path |> Path.dirname() do upload("#{uri.scheme}://#{uri.host}#{parent_path}/", cookie_credential) end end defp write(%HTTPoison.Response{status_code: 409}, destination_uri, _) do raise ArgumentError, message: "The path #{destination_uri} already exists. Please select another destination." end defp upload_to(%HTTPoison.Response{status_code: 303} = response, cookie_credential) do response.headers |> get_header("Location") |> HTTPoison.get!([{"Cookie", "CADC_SSO=\"#{cookie_credential}\""}]) |> upload_to(cookie_credential) end defp upload_to(%HTTPoison.Response{status_code: 200} = response, cookie_credential) do with transfer_url = Parser.all( Parser.from_string(response.body), "//vos:transfer/vos:protocol[@uri=\"ivo://ivoa.net/vospace/core#httpput\"]/vos:endpoint" ) |> extract_transfer_endpoint_url do # Dip into Hackney to allow streaming the body asynchronously. :hackney.request( "put", transfer_url, [{<<"Cookie">>, <<"CADC_SSO=\"#{cookie_credential}\"">>}], :stream, [] ) end end defp upload_to(response, _) do raise ArgumentError, message: response.body end defp extract_transfer_endpoint_url([]) do raise ArgumentError, message: "No service endpoints found to transfer to." end defp extract_transfer_endpoint_url(endpoints) do endpoints |> Enum.at(0) |> Parser.text() end defp to_xml(destination_uri) do with node_type = get_node_type(destination_uri) do cond do node_type == "ContainerNode" -> "" node_type == "DataNode" -> "" true -> raise ArgumentError, message: "No such supported Node Type: '#{node_type}'" end end end defp to_xml(destination_uri, transfer_url) do "#{destination_uri}pushToVoSpace#{ transfer_url }true" end defp get_node_type(uri) do cond do String.ends_with?(uri, "/") -> "ContainerNode" true -> "DataNode" end end defp get_header(headers, key) do headers |> Enum.filter(fn {k, _} -> k == key end) |> hd |> elem(1) end end