Module oracle::aq

Expand description

Oracle Advanced Queuing (available when aq_unstable feature is enabled.)

Warning: Any type in this module is unstable. It may be changed incompatibly by minor version upgrades.

Examples

Object type queue


// Create a queue
let objtype = conn.object_type("UDT_BOOK")?;
let mut queue = aq::Queue::<Object>::new(&conn, "BOOK_QUEUE", &objtype)?;

// Create a message
let mut payload = objtype.new_object()?;
payload.set("TITLE", &"Pebble in the Sky")?;
payload.set("AUTHORS", &"Isaac Asimov")?;
payload.set("PRICE", &17.0)?;
let mut msg = aq::MsgProps::<Object>::new(&conn)?;
msg.set_payload(&payload);

// Enqueue the message to the queue
queue.enqueue(&msg)?;

// Dequeue a message from the queue
let new_msg = queue.dequeue()?;
let new_payload = new_msg.payload()?;

// Compare message payloads.
assert_eq!(payload.get::<String>("TITLE")?, new_payload.get::<String>("TITLE")?);
assert_eq!(payload.get::<String>("AUTHORS")?, new_payload.get::<String>("AUTHORS")?);
assert_eq!(payload.get::<f32>("PRICE")?, new_payload.get::<f32>("PRICE")?);

RAW data queue


// Create a queue
let mut queue = aq::Queue::<[u8]>::new(&conn, "RAW_QUEUE", &())?;

// Create a message
let payload = b"\xde\xad\xbe\xef";
let mut msg = aq::MsgProps::<[u8]>::new(&conn)?;
msg.set_payload(payload.as_ref());

// Enqueue the message to the queue
queue.enqueue(&msg)?;

// Dequeue a message from the queue
let new_msg = queue.dequeue()?;
let new_payload = new_msg.payload()?; // returns Vec<u8>

// Compare message payloads.
assert_eq!(payload, new_payload.as_slice());

Enqueue and dequeue more than one message in one call


// Create a queue
let mut queue = aq::Queue::<[u8]>::new(&conn, "RAW_QUEUE", &())?;

// Create messages
let payloads = [b"\xde\xad\xbe\xef", b"\xba\xce\xba\x11"];
let mut messages = vec![];
for payload in &payloads {
    let mut msg = aq::MsgProps::<[u8]>::new(&conn)?;
    msg.set_payload(payload.as_ref())?;
    messages.push(msg);
}

// Enqueue the messages
queue.enqueue_many(&messages)?;

// Dequeue messages from the queue
let new_messages = queue.dequeue_many(10)?;

// Compare message payloads.
assert_eq!(new_messages.len(), 2);
assert_eq!(new_messages[0].payload()?, payloads[0]);
assert_eq!(new_messages[1].payload()?, payloads[1]);