Merge branch 'fix-547'

NEWS

@@ -16,6 +16,8 @@ What's new in psycopg 2.7.2
   (:ticket:`#544`).
 - Fixed `~connection.lobject()` keyword argument *lobject_factory*
   (:ticket:`#545`).
+- Fixed `~psycopg2.extras.ReplicationCursor.consume_stream()`
+  *keepalive_interval* argument (:ticket:`#547`).
 
 
 What's new in psycopg 2.7.1

psycopg/replication_cursor_type.c

@@ -85,7 +85,7 @@ psyco_repl_curs_consume_stream(replicationCursorObject *self,
     double keepalive_interval = 10;
     static char *kwlist[] = {"consume", "keepalive_interval", NULL};
 
-    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|id", kwlist,
+    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "O|d", kwlist,
                                      &consume, &keepalive_interval)) {
         return NULL;
     }

tests/test_replication.py

@@ -159,6 +159,29 @@ class ReplicationTest(ReplicationTestCase):
         # try with correct command
         cur.start_replication(slot_name=self.slot)
 
+    @skip_before_postgres(9, 4)  # slots require 9.4
+    @skip_repl_if_green
+    def test_keepalive(self):
+        conn = self.repl_connect(connection_factory=LogicalReplicationConnection)
+        if conn is None:
+            return
+
+        cur = conn.cursor()
+
+        self.create_replication_slot(cur, output_plugin='test_decoding')
+
+        self.make_replication_events()
+
+        cur.start_replication(self.slot)
+
+        def consume(msg):
+            raise StopReplication()
+
+        self.assertRaises(StopReplication,
+            cur.consume_stream, consume, keepalive_interval=2)
+
+        conn.close()
+
     @skip_before_postgres(9, 4)     # slots require 9.4
     @skip_repl_if_green
     def test_stop_replication(self):