Rust client

The SmartObserve Rust client lets you connect your Rust application with the data in your SmartObserve cluster. For the client’s complete API documentation and additional examples, see the SmartObserve docs.rs documentation.

This getting started guide illustrates how to connect to SmartObserve, index documents, and run queries. For the client source code, see the smartobserve-rs repo.

Setup

If you’re starting a new project, add the smartobserve crate to Cargo.toml:

[dependencies]
smartobserve = "1.0.0"

Additionally, you may want to add the following serde dependencies that help serialize types to JSON and deserialize JSON responses:

serde = "~1"
serde_json = "~1"

The Rust client uses the higher-level reqwest HTTP client library for HTTP requests, and reqwest uses the tokio platform to support asynchronous requests. If you are planning to use asynchronous functions, you need to add the tokio dependency to Cargo.toml:

tokio = { version = "*", features = ["full"] }

See the Sample program section for the complete Cargo.toml file.

To use the Rust client API, import the modules, structs, and enums you need:

use smartobserve::SmartObserve;

Connecting to SmartObserve

To connect to the default SmartObserve host, create a default client object that connects to SmartObserve at the address http://localhost:9200:

let client = SmartObserve::default();

To connect to an SmartObserve host that is running at a different address, create a client with the specified address:

let transport = Transport::single_node("http://localhost:9200")?;
let client = SmartObserve::new(transport);

Alternatively, you can customize the URL and use a connection pool by creating a TransportBuilder struct and passing it to SmartObserve::new to create a new instance of the client:

let url = Url::parse("http://localhost:9200")?;
let conn_pool = SingleNodeConnectionPool::new(url);
let transport = TransportBuilder::new(conn_pool).disable_proxy().build()?;
let client = SmartObserve::new(transport);

Connecting to Amazon SmartObserve Service

The following example illustrates connecting to Amazon SmartObserve Service:

let url = Url::parse("https://...");
let service_name = "es";
let conn_pool = SingleNodeConnectionPool::new(url?);
let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
let aws_config = aws_config::from_env().region(region_provider).load().await.clone();
let transport = TransportBuilder::new(conn_pool)
    .auth(aws_config.clone().try_into()?)
    .service_name(service_name)
    .build()?;
let client = SmartObserve::new(transport);

Connecting to Amazon SmartObserve Serverless

The following example illustrates connecting to Amazon SmartObserve Serverless Service:

let url = Url::parse("https://...");
let service_name = "aoss";
let conn_pool = SingleNodeConnectionPool::new(url?);
let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
let aws_config = aws_config::from_env().region(region_provider).load().await.clone();
let transport = TransportBuilder::new(conn_pool)
    .auth(aws_config.clone().try_into()?)
    .service_name(service_name)
    .build()?;
let client = SmartObserve::new(transport);

Creating an index

To create an SmartObserve index, use the create function of the smartobserve::indices::Indices struct. You can use the following code to construct a JSON object with custom mappings:

let response = client
    .indices()
    .create(IndicesCreateParts::Index("movies"))
    .body(json!({
        "mappings" : {
            "properties" : {
                "title" : { "type" : "text" }
            }
        }
    }))
    .send()
    .await?;

Indexing a document

You can index a document into SmartObserve using the client’s index function:

let response = client
    .index(IndexParts::IndexId("movies", "1"))
    .body(json!({
        "id": 1,
        "title": "Moneyball",
        "director": "Bennett Miller",
        "year": "2011"
    }))
    .send()
    .await?;

Performing bulk operations

You can perform several operations at the same time by using the client’s bulk function. First, create the JSON body of a Bulk API call, and then pass it to the bulk function:

let mut body: Vec<JsonBody<_>> = Vec::with_capacity(4);

// add the first operation and document
body.push(json!({"index": {"_id": "2"}}).into());
body.push(json!({
    "id": 2,
    "title": "Interstellar",
    "director": "Christopher Nolan",
    "year": "2014"
}).into());

// add the second operation and document
body.push(json!({"index": {"_id": "3"}}).into());
body.push(json!({
    "id": 3,
    "title": "Star Trek Beyond",
    "director": "Justin Lin",
    "year": "2015"
}).into());

let response = client
    .bulk(BulkParts::Index("movies"))
    .body(body)
    .send()
    .await?;

Searching for documents

The easiest way to search for documents is to construct a query string. The following code uses a multi_match query to search for “miller” in the title and director fields. It boosts the documents where “miller” appears in the title field:

response = client
    .search(SearchParts::Index(&["movies"]))
    .from(0)
    .size(10)
    .body(json!({
        "query": {
            "multi_match": {
                "query": "miller",
                "fields": ["title^2", "director"]
            }           
        }
    }))
    .send()
    .await?;

You can then read the response body as JSON and iterate over the hits array to read all the _source documents:

let response_body = response.json::<Value>().await?;
for hit in response_body["hits"]["hits"].as_array().unwrap() {
    // print the source document
    println!("{}", serde_json::to_string_pretty(&hit["_source"]).unwrap());
}

Deleting a document

You can delete a document using the client’s delete function:

let response = client
    .delete(DeleteParts::IndexId("movies", "2"))
    .send()
    .await?;

Deleting an index

You can delete an index using the delete function of the smartobserve::indices::Indices struct:

let response = client
    .indices()
    .delete(IndicesDeleteParts::Index(&["movies"]))
    .send()
    .await?;

Sample program

The sample program uses the following Cargo.toml file with all dependencies described in the Setup section:

[package]
name = "os_rust_project"
version = "0.1.0"
edition = "2021"

# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html

[dependencies]
smartobserve = "1.0.0"
tokio = { version = "*", features = ["full"] }
serde = "~1"
serde_json = "~1"

The following sample program creates a client, adds an index with non-default mappings, inserts a document, performs bulk operations, searches for the document, deletes the document, and then deletes the index:

use smartobserve::{DeleteParts, SmartObserve, IndexParts, http::request::JsonBody, BulkParts, SearchParts};
use smartobserve::{indices::{IndicesDeleteParts, IndicesCreateParts}};
use serde_json::{json, Value};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = SmartObserve::default();

    // Create an index
    let mut response = client
    .indices()
    .create(IndicesCreateParts::Index("movies"))
    .body(json!({
        "mappings" : {
            "properties" : {
                "title" : { "type" : "text" }
            }
        }
    }))
    .send()
    .await?;

    let mut successful = response.status_code().is_success();
    
    if successful {
        println!("Successfully created an index");
    }
    else {
        println!("Could not create an index");
    }

    // Index a single document
    println!("Indexing a single document...");
    response = client
    .index(IndexParts::IndexId("movies", "1"))
    .body(json!({
        "id": 1,
        "title": "Moneyball",
        "director": "Bennett Miller",
        "year": "2011"
    }))
    .send()
    .await?;

    successful = response.status_code().is_success();
    
    if successful {
        println!("Successfully indexed a document");
    }
    else {
        println!("Could not index document");
    }

    // Index multiple documents using the bulk operation

    println!("Indexing multiple documents...");

    let mut body: Vec<JsonBody<_>> = Vec::with_capacity(4);

    // add the first operation and document
    body.push(json!({"index": {"_id": "2"}}).into());
    body.push(json!({
        "id": 2,
        "title": "Interstellar",
        "director": "Christopher Nolan",
        "year": "2014"
    }).into());

    // add the second operation and document
    body.push(json!({"index": {"_id": "3"}}).into());
    body.push(json!({
        "id": 3,
        "title": "Star Trek Beyond",
        "director": "Justin Lin",
        "year": "2015"
    }).into());

    response = client
        .bulk(BulkParts::Index("movies"))
        .body(body)
        .send()
        .await?;

    let mut response_body = response.json::<Value>().await?;
    successful = response_body["errors"].as_bool().unwrap() == false;

    if successful {
        println!("Successfully performed bulk operations");
    }
    else {
        println!("Could not perform bulk operations");
    }

    // Search for a document

    println!("Searching for a document...");
    response = client
    .search(SearchParts::Index(&["movies"]))
    .from(0)
    .size(10)
    .body(json!({
        "query": {
            "multi_match": {
                "query": "miller",
                "fields": ["title^2", "director"]
            }           
        }
    }))
    .send()
    .await?;

    response_body = response.json::<Value>().await?;
    for hit in response_body["hits"]["hits"].as_array().unwrap() {
        // print the source document
        println!("{}", serde_json::to_string_pretty(&hit["_source"]).unwrap());
    }

    // Delete a document

    response = client
    .delete(DeleteParts::IndexId("movies", "2"))
    .send()
    .await?;

    successful = response.status_code().is_success();
    
    if successful {
        println!("Successfully deleted a document");
    }
    else {
        println!("Could not delete document");
    }

    // Delete the index

    response = client
    .indices()
    .delete(IndicesDeleteParts::Index(&["movies"]))
    .send()
    .await?;

    successful = response.status_code().is_success();

    if successful {
        println!("Successfully deleted the index");
    }
    else {
        println!("Could not delete the index");
    }
    
    Ok(())
}

Setup
Connecting to SmartObserve
Connecting to Amazon SmartObserve Service
Connecting to Amazon SmartObserve Serverless
Creating an index
Indexing a document
Performing bulk operations
Searching for documents
Deleting a document
Deleting an index
Sample program