From 248e653c9e2ec2dfc0b1eac9561981b34a637f9d Mon Sep 17 00:00:00 2001
From: Daniele Varrazzo <daniele.varrazzo@gmail.com>
Date: Wed, 19 Apr 2017 01:01:59 +0100
Subject: [PATCH] Fixed args parsing in ReplicationCursor.consume_stream()

Close #547.
---
 NEWS                              |  2 ++
 psycopg/replication_cursor_type.c |  2 +-
 tests/test_replication.py         | 23 +++++++++++++++++++++++
 3 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/NEWS b/NEWS
index a0399a15..55b0b224 100644
--- a/NEWS
+++ b/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
diff --git a/psycopg/replication_cursor_type.c b/psycopg/replication_cursor_type.c
index d66bec36..d00b31ce 100644
--- a/psycopg/replication_cursor_type.c
+++ b/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;
     }
diff --git a/tests/test_replication.py b/tests/test_replication.py
index 0aed578f..182bff21 100755
--- a/tests/test_replication.py
+++ b/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):