Struct df_kafka::Kafka

pub struct Kafka { /* private fields */ }

Implementations

impl Kafka

pub fn connect(config: JsonValue) -> Self

链接

• config 配置

{
  "default": "mac",
  "connections": {
    "mac": {
      "brokers": [
        "0.0.0.0:9092"
      ],
      "fetch_max": 900000000,
      "retry_max": 900000000
    }
  }
}

Examples found in repository

examples/test.rs (line 27)

fn main() {
    env_logger::init();

    let mut root_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
    root_path = root_path.join("..");
    let conf_path = root_path.join("config");
    let conf_path = conf_path.join("kafka.json");

    let conf = match fs::read_to_string(conf_path) {
        Ok(content) => {
            if content == "" {
                object! {}
            } else {
                json::parse(content.as_str()).unwrap()
            }
        }
        Err(_) => {
            object! {}
        }
    };
    let kafka = Kafka::connect(conf);

    let data = kafka.set("tests", "", "112".into());
    info!(">>>>>>{}", data);
    let data = kafka.set("tests", "", "113".into());
    info!(">>>>>>{}", data);

    let data = kafka.get("tests", "", false);
    info!(">>>>>>{}", data);
}

pub fn add_config(&mut self, connection: JsonValue) -> &mut Self

增加配置

pub fn switch_config(&mut self, default: &str) -> &mut Self

切换配置

pub fn set(&self, topic: &str, key: &str, data: JsonValue) -> bool

生产

Examples found in repository

examples/test.rs (line 29)

fn main() {
    env_logger::init();

    let mut root_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
    root_path = root_path.join("..");
    let conf_path = root_path.join("config");
    let conf_path = conf_path.join("kafka.json");

    let conf = match fs::read_to_string(conf_path) {
        Ok(content) => {
            if content == "" {
                object! {}
            } else {
                json::parse(content.as_str()).unwrap()
            }
        }
        Err(_) => {
            object! {}
        }
    };
    let kafka = Kafka::connect(conf);

    let data = kafka.set("tests", "", "112".into());
    info!(">>>>>>{}", data);
    let data = kafka.set("tests", "", "113".into());
    info!(">>>>>>{}", data);

    let data = kafka.get("tests", "", false);
    info!(">>>>>>{}", data);
}

pub fn get(&self, topic: &str, key: &str, only: bool) -> JsonValue

消费

Examples found in repository

examples/test.rs (line 34)

fn main() {
    env_logger::init();

    let mut root_path = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
    root_path = root_path.join("..");
    let conf_path = root_path.join("config");
    let conf_path = conf_path.join("kafka.json");

    let conf = match fs::read_to_string(conf_path) {
        Ok(content) => {
            if content == "" {
                object! {}
            } else {
                json::parse(content.as_str()).unwrap()
            }
        }
        Err(_) => {
            object! {}
        }
    };
    let kafka = Kafka::connect(conf);

    let data = kafka.set("tests", "", "112".into());
    info!(">>>>>>{}", data);
    let data = kafka.set("tests", "", "113".into());
    info!(">>>>>>{}", data);

    let data = kafka.get("tests", "", false);
    info!(">>>>>>{}", data);
}

pub fn get_client(&self, topic: &str, fun: fn(str: String) -> bool)

消费 监听消息不退出