Fixed args parsing in ReplicationCursor.consume_stream()

Close #547.

NEWS

@@ -12,6 +12,8 @@ What's new in psycopg 2.7.2
   string is specified as an URI (:ticket:`#528`).
 - Return objects with timezone parsing "infinity" :sql:`timestamptz`
   (:ticket:`#536`).
+- 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):