Flush ascii get, gets, incr, decr, flush_all and binary gets requests

src/protocol/ascii.rs

@@ -144,6 +144,7 @@ impl ProtocolTrait for AsciiProtocol<Stream> {
 
     fn flush(&mut self) -> Result<(), MemcacheError> {
         write!(self.reader.get_mut(), "flush_all\r\n")?;
+        self.reader.get_mut().flush()?;
         self.parse_ok_response()
     }
 
@@ -161,6 +162,7 @@ impl ProtocolTrait for AsciiProtocol<Stream> {
         };
 
         write!(self.reader.get_mut(), "{} {}\r\n", command, key)?;
+        self.reader.get_mut().flush()?;
 
         if let Some((k, v)) = self.parse_get_response(has_cas)? {
             if k != key {
@@ -179,6 +181,7 @@ impl ProtocolTrait for AsciiProtocol<Stream> {
 
     fn gets<V: FromMemcacheValueExt>(&mut self, keys: &[&str]) -> Result<HashMap<String, V>, MemcacheError> {
         write!(self.reader.get_mut(), "gets {}\r\n", keys.join(" "))?;
+        self.reader.get_mut().flush()?;
 
         let mut result: HashMap<String, V> = HashMap::with_capacity(keys.len());
         // there will be atmost keys.len() "VALUE <...>" responses and one END response
@@ -273,11 +276,13 @@ impl ProtocolTrait for AsciiProtocol<Stream> {
 
     fn increment(&mut self, key: &str, amount: u64) -> Result<u64, MemcacheError> {
         write!(self.reader.get_mut(), "incr {} {}\r\n", key, amount)?;
+        self.reader.get_mut().flush()?;
         self.parse_u64_response()
     }
 
     fn decrement(&mut self, key: &str, amount: u64) -> Result<u64, MemcacheError> {
         write!(self.reader.get_mut(), "decr {} {}\r\n", key, amount)?;
+        self.reader.get_mut().flush()?;
         self.parse_u64_response()
     }
 

src/protocol/binary.rs

@@ -99,6 +99,7 @@ impl ProtocolTrait for BinaryProtocol {
             ..Default::default()
         };
         noop_request_header.write(&mut self.stream)?;
+        self.stream.flush()?;
         return binary_packet::parse_gets_response(&mut self.stream, keys.len());
     }
 

tests/tests.rs

@@ -141,7 +141,8 @@ fn udp_test() {
     assert_eq!(client.touch("foooo", 123).unwrap(), false);
     assert_eq!(client.touch("fooo", 12345).unwrap(), true);
 
-    // gets is not supported for udp
+    // memcached sends every response as its own datagram sequence and
+    // UdpStream only reads one, so gets can not work over udp
     let value: Result<std::collections::HashMap<String, String>, _> = client.gets(&["foo", "fooo"]);
     assert_eq!(value.is_ok(), false);
 
@@ -200,6 +201,23 @@ fn udp_test() {
     for i in 0..10 {
         handles[i].take().unwrap().join().unwrap();
     }
+
+    let client = memcache::Client::connect("memcache+udp://localhost:22345?protocol=ascii").unwrap();
+
+    client.set("ascii_udp", "bar", 0).unwrap();
+    let value: Option<String> = client.get("ascii_udp").unwrap();
+    assert_eq!(value, Some(String::from("bar")));
+
+    client.set("ascii_udp_counter", 41, 0).unwrap();
+    assert_eq!(client.increment("ascii_udp_counter", 2).unwrap(), 43);
+    assert_eq!(client.decrement("ascii_udp_counter", 1).unwrap(), 42);
+
+    let values: Result<std::collections::HashMap<String, String>, _> = client.gets(&["ascii_udp", "ascii_udp_counter"]);
+    assert_eq!(values.is_ok(), false);
+
+    client.flush().unwrap();
+    let value: Option<String> = client.get("ascii_udp").unwrap();
+    assert_eq!(value, None);
 }
 
 #[test]