Example of using packet-stream-rs

main.rs

#![feature(try_from)]
#![feature(ip_constructors)]

extern crate futures;
extern crate tokio;
extern crate tokio_io;
extern crate sodiumoxide;
extern crate secret_stream;
extern crate ssb_common;
extern crate ssb_keyfile;
extern crate packet_stream;

use std::net::{SocketAddr, Ipv6Addr};
use std::convert::TryInto;

use tokio::net::TcpStream;
use secret_stream::Client;
use sodiumoxide::crypto::box_;
use futures::prelude::*;
use futures::future::{ok, poll_fn};
use tokio_io::AsyncRead;
use packet_stream::*;
use ssb_common::*;
use ssb_keyfile::load_or_create_keys;

fn main() {
    // Should always be called before using any of the crypto stuff: It makes
    // things threadsafe and faster.
    sodiumoxide::init();

    // Read keys from the secret file from the ssb directory. Takes the
    // `ssb_appname` environment variable into account.
    let (pk, sk) = load_or_create_keys().unwrap();
    let pk = pk.try_into().unwrap();
    let sk = sk.try_into().unwrap();
    let (ephemeral_pk, ephemeral_sk) = box_::gen_keypair();

    let addr = SocketAddr::new(Ipv6Addr::localhost().into(), DEFAULT_TCP_PORT);
    TcpStream::connect(&addr)
        .and_then(|tcp| {
            // Performs a secret-handshake and yields an encrypted duplex connection.
            Client::new(tcp,
                        &MAINNET_IDENTIFIER,
                        &pk,
                        &sk,
                        &ephemeral_pk,
                        &ephemeral_sk,
                        &pk)
                    .map_err(|(err, _)| err)
        })
        .and_then(|connection| {
            // Split the connection into its halves and create a packet-stream.
            let (read, write) = connection.unwrap().split();
            let (ps_in, mut ps_out) = packet_stream(read, write);

            let (send_request, response) = ps_out
                .request(&b"{\"name\":[\"whoami\"],\"type\":\"sync\",\"args\":[]}"[..],
                         PsPacketType::Json);

            let use_response = response.map(|res| {
                                                println!("{:?}",
                                                         String::from_utf8(res.into_data()));
                                                ()
                                            });

            // execute the `whoami` rpc
            send_request
                // while dealing with the response in parallel
                .join(use_response)
                // close this end of the packet-stream afterwards
                .and_then(|_| poll_fn(move || ps_out.close()))
                // meanwhile consume the incoming packets and stop execution
                // if the server closes the connection
                .select(ps_in.for_each(|_| ok(())))
                // keep the typechecker happy
                .map(|(item, _)| item)
                .map_err(|(err, _)| err)
        })
        // Actually run the computations.
        .wait()
        .unwrap();
}