C++/boost, rust, elixir でTCP通信

#動機
MPD(music player daemon)のクライアントを作れないかなぁとか思いたった。
そこで、ポストC++で有名なrustで書いてみていた。しかし、MPDの情報を取得するライブラリがc++とpythonでは提供されていたが、他では無いことがわかった。
なので、直接MPDにTCP通信でクエリを投げて情報を取り出そうと思い色々実験してみた。

通信はすべて同期通信で行っている。

TCP通信

TCP通信とはデータ通信をするときのプロトコルのこと。
UDPプロトコルと違って、再送処理を行うための仕組みを含んでいるので一般に信頼性が高いと言われている。

#ソース

Rust

main.rs

use std::string::String;
use std::io::prelude::*;
use std::collections::HashMap;
use std::net::{TcpStream,SocketAddrV4,Ipv4Addr};

fn main() {
    let mut mpd: TcpStream = get_mpd_socket(Ipv4Addr::new(127,0,0,1), 6600);
    let current: HashMap<String, String>  = currentsong(&mut mpd);
    for key in current.keys() {
        println!("{}: {}", key, current.get(key).unwrap());
    }
}

fn get_mpd_socket(addr: Ipv4Addr, port: u16) -> TcpStream {
    let mut mpd: TcpStream = TcpStream::connect(SocketAddrV4::new(addr, port)).unwrap();

    // C++みたいに指定した文字列が出てくるまで読み込むみたいなのがなかったので、
    // とりあえず、オーバーして読み込んでそれ以上を読み込み始めたら1usでタイムアウトさせている.
    let _ = mpd.set_read_timeout(Some(std::time::Duration::new(0,1)));

    // MPDはコネクトした直後に"MPD $(mpd_version)"を返すので、それを捨てている.
    let mut buf: String = String::new();
    let _ = mpd.read_to_string(&mut buf);

    return mpd;
}

// MPDの戻り値が
// Title: **
// Album: **
// てな感じなのでHashMapに変換してみた.
fn to_map(data: String) -> HashMap<String, String> {
    // "Title:**\nAlbum: **\n" to ["Title: **", "Album: **"]
    let split_tmp: Vec<&str> = data.split("\n").collect();
    // ["Title: **", "Album: **"] to [["Title", "**"], ["Album", "**"]]
    let mut split_list: Vec<Vec<&str>> = split_tmp
                            .into_iter()
                            .map(|x| x.split(": ").collect())
                            .collect();

    // リクエストに対するMPDの返答の一番最後は"OK\n"なので、それをpopして捨てる.
    let _ = split_list.pop();
    let _ = split_list.pop();
    let mut map: HashMap<String, String> = HashMap::new();

    // {"Title" => "**", "Album" => "**"}
    for i in 0..split_list.len() {
        map.insert(split_list[i][0].to_string(), split_list[i][1].to_string());
    }

    return map;
}

fn currentsong(mpd: &mut TcpStream) -> HashMap<String, String> {
    let mut buf: String = String::new();

    let _ = mpd.write(b"currentsong\n");
    let _ = mpd.read_to_string(&mut buf);

    return to_map(buf);
}

C++

• compile: g++ tcp.cpp -l boost_system -l pthread

tcp.cpp

#include <stdio.h>
#include <iostream>
#include <string>
#include <boost/asio.hpp>
#include <unistd.h>

int main(void) {
  // 各々のOSのI/O制御への橋渡しをするクラスらしい. 詳しくはわからん
  boost::asio::io_service io;

  boost::asio::ip::tcp::socket sock(io);

  boost::asio::ip::tcp::endpoint endpoint = boost::asio::ip::tcp::endpoint{
    boost::asio::ip::address::from_string("127.0.0.1"),   // IPv4 address
    6600                                                  // port
  };

  sock.connect(endpoint);

  boost::asio::streambuf request;
  std::ostream request_ostream(&request);
  request_ostream << "currentsong\n";  // MPDで現在再生中の音楽の情報をリクエスト

  boost::system::error_code error;

  // MPDにデータをwrite
  boost::asio::write(sock,request);

  boost::asio::streambuf buffer;

  // MPDが送りつけてくるデータの最後の行は成功なら "OK\n"なのでそこまで読み込む
  boost::asio::read_until(sock,buffer,"OK\n");
  if(error && error != boost::asio::error::eof){
    std::cout << "receive failed" << std::endl;
  } else {
    std::cout << &buffer;
  }

  return 0;
}

Elixir

tcp.ex

defmodule MPDQuery do
  # MPDとのソケットを戻す
  defp connected_to_mpd(port) do
    {:ok, socket} = :gen_tcp.connect('localhost', port, [:binary, active: false])
    # 接続時の "MPD $(mpd_version)" を捨てる.
    _ = :gen_tcp.recv(socket, 0)

    # RECVで使うバッファのサイズ指定 now(10^8)
    # elixirはreceiveするbyteを指定するか,recbufを上限にデータを取得するが
    # recbufの初期値が小さいのでデータを取得しきれない場合がでてくるので
    # ここではとりあえず膨大な値を暫定的に入れている。
    :ok = :inet.setopts(socket, recbuf: 100000000)
    socket
  end

  def to_map(data) do
    Regex.split(~r/\n/, data)
      |> Enum.map(&(Regex.split(~r/:\s/, &1)))
      |> Enum.map(&(%{String.to_atom(Enum.at(&1,0)) => Enum.at(&1,1)}))
  end

  defp map_in_list_merge(list) do
    if list |> Enum.count > 1 do
      list
        |> Enum.at(0)
        |> Map.merge(map_in_list_merge(list |> Enum.drop(1)))
    else
      list |> Enum.at(0)
    end
  end


  def current do
    cmd_do("currentsong \n")
      |> map_in_list_merge
  end

  defp cmd_do(cmd) do
    sock = connected_to_mpd(6600)

    :ok = :gen_tcp.send(sock, cmd)
    :timer.sleep(10)
    {:ok, msg} = :gen_tcp.recv(sock, 0)

    :ok = :gen_tcp.close(sock)

    # 受信したデータの最後の "OK\n"を削除
    Enum.join(for <<c::utf8 <- msg>>, do: <<c::utf8>>)
      |> to_map
      |> List.delete(%{"": nil})
      |> List.delete(%{OK: nil})
  end
end

IO.inspect MPDQuery.current

まとめ

• C++では、特定の文字列が出現するまでデータを読み込む(read_until)があったので、データの受信では苦労しなかった。

• Rustでは、送られてきたデータ全てを受け取る(EOFが出現するまで)ことができた。が、?
  なぜか、タイムアウトまで待たないとデータ受信が完了せず、やたらと時間がかかってしまうので、タイムアウトを1uにして無理やり解決した。

• Elixirは、そのままrecv/2で全てのデータを取得することができるが、そのまま実行すると、受信できる量が少ないようだった。
  原因はわからないが、recbufという、データを受信するときに使うバッファの容量を指定するオプションで多めに指定すると取得できたので、とりあえず、それでやった。
  ちなみに、getopts/2でソケットのオプションを取得できるので調べたところ、

初期状態
{:ok, [recbuf, 1061808]}
setopts(socket, recbuf: 100000000) 実行後
{:ok, [recbuf, 425984]}

で、もはやバッファ減ってたので、原因は別のところにあると思うが、どうしたらいいのかわからぬ
教えてエロい人