Adds federation to Wikibase server

README.md

@@ -76,11 +76,13 @@ firefox http://localhost:7878
 curl http://localhost:7878 -H 'Content-Type: application/x-turtle' -d@./data.ttl
 
 # Make a query
-curl -H 'Accept: application/sparql-results+json' 'http://localhost:7878/query?query=SELECT%20*%20%7B%20%3Fs%20%3Fp%20%3Fo%20%7D%20LIMIT%2010'
+curl -X POST -H 'Accept: application/sparql-results+json' -H 'Content-Type: application/sparql-query' --data 'SELECT * WHERE { ?s ?p ?o } LIMIT 10' http://localhost:7878/query
 ```
 
 You could easily build your own Docker image by running `docker build -t oxigraph server -f server/Dockerfile .` from the root directory.
 
+## Run the Wikibase server
+
 ### Build
 
 You need to have [a recent stable version of Rust and Cargo installed](https://www.rust-lang.org/tools/install).
@@ -92,7 +94,7 @@ It will create a fat binary in `target/release/oxigraph_wikibase`.
 
 To start a server that is synchronized with [test.wikidata.org](https://test.wikidata.org) you should run:
 ```bash
-./oxigraph_wikibase --mediawiki-api=https://test.wikidata.org/w/api.php --mediawiki-base-url=https://test.wikidata.org/wiki/ --namespaces=0,120 --file=test.wikidata
+./oxigraph_wikibase --mediawiki-api https://test.wikidata.org/w/api.php --mediawiki-base-url https://test.wikidata.org/wiki/ --namespaces 0,120 --file test.wikidata
 ```
 
 It creates a SPARQL endpoint listening to `localhost:7878/query` that could be queried just like Blazegraph.
@@ -103,6 +105,12 @@ The configuration parameters are:
 * `namespaces` The ids of the Wikibase namespaces to synchronize with, separated by `,`.
 * `file` Path of where Oxigraph should store its data.
 
+
+You can then access it from your machine on port `7878`. No GUI is provided.
+```sh
+# Make a query
+curl -X POST -H 'Accept: application/sparql-results+json' -H 'Content-Type: application/sparql-query' --data 'SELECT * WHERE { ?s ?p ?o } LIMIT 10' http://localhost:7878/query
+```
 ### Using a Docker image
 
 #### Display the help menu

server/src/main.rs

@@ -379,6 +379,7 @@ impl From<Error> for HttpServiceError {
         Self { inner }
     }
 }
+
 #[cfg(test)]
 mod tests {
     use crate::handle_request;

wikibase/src/main.rs

@@ -14,11 +14,18 @@ use argh::FromArgs;
 use async_std::future::Future;
 use async_std::net::{TcpListener, TcpStream};
 use async_std::prelude::*;
-use async_std::task::{spawn, spawn_blocking};
+use async_std::task::{block_on, spawn, spawn_blocking};
-use http_types::{headers, Body, Error, Method, Mime, Request, Response, Result, StatusCode};
+use http_client::h1::H1Client;
+use http_client::HttpClient;
+use http_types::{
+    format_err, headers, Body, Error, Method, Mime, Request, Response, Result, StatusCode, Url,
+};
 use oxigraph::io::GraphFormat;
-use oxigraph::sparql::{Query, QueryOptions, QueryResult, QueryResultFormat};
+use oxigraph::model::NamedNode;
+use oxigraph::sparql::{Query, QueryOptions, QueryResult, QueryResultFormat, ServiceHandler};
 use oxigraph::RocksDbStore;
+use std::fmt;
+use std::io::Cursor;
 use std::str::FromStr;
 use std::time::Duration;
 use url::form_urlencoded;
@@ -180,7 +187,10 @@ async fn evaluate_sparql_query(
             e.set_status(StatusCode::BadRequest);
             e
         })?;
-        let results = store.query(query, QueryOptions::default())?;
+        let options = QueryOptions::default()
+            .with_default_graph_as_union()
+            .with_service_handler(HttpService::default());
+        let results = store.query(query, options)?;
         if let QueryResult::Graph(_) = results {
             let format = content_negotiation(
                 request,
@@ -292,3 +302,70 @@ fn content_negotiation<F>(
     parse(result.essence())
         .ok_or_else(|| Error::from_str(StatusCode::InternalServerError, "Unknown mime type"))
 }
+
+#[derive(Default)]
+struct HttpService {
+    client: H1Client,
+}
+
+impl ServiceHandler for HttpService {
+    type Error = HttpServiceError;
+
+    fn handle(
+        &self,
+        service_name: NamedNode,
+        query: Query,
+    ) -> std::result::Result<QueryResult, HttpServiceError> {
+        let mut request = Request::new(
+            Method::Post,
+            Url::parse(service_name.as_str()).map_err(Error::from)?,
+        );
+        request.append_header(headers::USER_AGENT, SERVER);
+        request.append_header(headers::CONTENT_TYPE, "application/sparql-query");
+        request.append_header(headers::ACCEPT, "application/sparql-results+xml");
+        request.set_body(query.to_string());
+
+        //TODO: response streaming
+        let response: Result<(Option<Mime>, Vec<u8>)> = block_on(async {
+            let mut response = self.client.send(request).await?;
+            Ok((response.content_type(), response.body_bytes().await?))
+        });
+        let (content_type, data) = response?;
+
+        let syntax = if let Some(content_type) = content_type {
+            QueryResultFormat::from_media_type(content_type.essence()).ok_or_else(|| {
+                format_err!(
+                    "Unexpected federated query result type from {}: {}",
+                    service_name,
+                    content_type
+                )
+            })?
+        } else {
+            QueryResultFormat::Xml
+        };
+        Ok(QueryResult::read(Cursor::new(data), syntax).map_err(Error::from)?)
+    }
+}
+
+#[derive(Debug)]
+struct HttpServiceError {
+    inner: Error,
+}
+
+impl fmt::Display for HttpServiceError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.inner.fmt(f)
+    }
+}
+
+impl std::error::Error for HttpServiceError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        Some(self.inner.as_ref())
+    }
+}
+
+impl From<Error> for HttpServiceError {
+    fn from(inner: Error) -> Self {
+        Self { inner }
+    }
+}