1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

// async_client.rs
// This file is part of the Eclipse Paho MQTT Rust Client library.
//

/*******************************************************************************
 * Copyright (c) 2017 Frank Pagliughi <fpagliughi@mindspring.com>
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *    http://www.eclipse.org/legal/epl-v10.html
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Frank Pagliughi - initial implementation and documentation
 *******************************************************************************/

use std::time::Duration;
use std::sync::mpsc::{Receiver};
//use std::sync::mpsc;

use async_client::{AsyncClient};
use create_options::{CreateOptions};
use connect_options::{ConnectOptions};
use disconnect_options::{DisconnectOptions};
use message::{Message};
use errors::{MqttResult, /*MqttError, ErrorKind*/};

/////////////////////////////////////////////////////////////////////////////
// Client

pub struct Client {
	cli: Box<AsyncClient>,
	timeout: Duration,
}

impl Client {
	pub fn new<T>(opts: T) -> MqttResult<Client>
		where T: Into<CreateOptions> 
	{
		let async_cli = AsyncClient::new(opts)?;
		let cli = Client {
			cli: Box::new(async_cli),
			timeout: Duration::from_secs(5*60),
		};
		//cli.start_consuming();
		Ok(cli)
	}

	pub fn connect<T: Into<Option<ConnectOptions>>>(&self, opt_opts:T) -> MqttResult<()> {
		self.cli.connect(opt_opts).wait_for(self.timeout)
	}

	pub fn disconnect<T: Into<Option<DisconnectOptions>>>(&self, opt_opts:T) -> MqttResult<()> {
		self.cli.disconnect(opt_opts).wait_for(self.timeout)
	}

	pub fn disconnect_after(&self, timeout: Duration) -> MqttResult<()> {
		self.cli.disconnect_after(timeout).wait_for(self.timeout)
	}

	pub fn reconnect(&self) -> MqttResult<()> {
		self.cli.reconnect().wait_for(self.timeout)
	}

	pub fn is_connected(&self) -> bool {
		self.cli.is_connected()
	}

	pub fn publish(&self, msg: Message) -> MqttResult<()> {
		self.cli.publish(msg).wait_for(self.timeout)
	}

	pub fn subscribe(&self, topic: &str, qos: i32) -> MqttResult<()> {
		self.cli.subscribe(topic, qos).wait_for(self.timeout)
	}

	pub fn subscribe_many(&self, topics: Vec<String>, qos: Vec<i32>) -> MqttResult<()> {
		self.cli.subscribe_many(topics, qos).wait_for(self.timeout)
	}

	pub fn unsubscribe(&self, topic: &str) -> MqttResult<()> {
		self.cli.unsubscribe(topic).wait_for(self.timeout)
	}

	pub fn unsubscribe_many(&self, topics: Vec<String>) -> MqttResult<()> {
		self.cli.unsubscribe_many(topics).wait_for(self.timeout)
	}

	pub fn start_consuming(&mut self) -> Receiver<Message> {
		self.cli.start_consuming()
	}
}