Stopping logical replication protocol

1. When WAL consumer catchup master and change his state to streaming, not available normally complete replication by send CopyDone message until will not generate/create new WAL record. It occurs because logical decoding located in WalSndWaitForWal until will not available next WAL record, and it method receive message from consumer even reply on CopyDone with CopyDone but ignore exit from loop and we can wait many times waiting CommandStatus & ReadyForQuery packages on consumer.
2. Logical decoding ignore message from consumer during decoding and writing transaction in socket(WalSndWriteData). It affect long transactions with many changes. Because for example if we start decoding transaction that insert 1 million records and after consume 1% of it date we
   decide stop replication, it will be not available until whole million record will not send to consumer. And I see the following problems because of this:

• It will generate many not necessary network traffic.
• Not available normally stop logical replication, because consumer will fail with timeout and it broke scenario when consumer put data to external system asynchronously and by success callback send feedback to master about flushed LSN, but by fail callback stop replication, and restart it from the last successfully sent LSN to external system, because slot will be busy and it will increase latency for streaming system.
• Consumer can send keepalive message with required reply flag to master that right now decoding and sending huge transaction, and after some time disconnect master because because reply on keepalive will not get. I faced with it problem during research  bottledwater-pg extension that fail each time during receive transaction in that was modify 1 million record(restart_lsn was before huge transaction so extension fail again and again) disconnect master because not keep keepalive package too long.

I prepare small patch that fix problems describe below. Now WalSndWriteData first check message from consumer and during decode transaction check that replication still active. KeppAlive message now not send if was get CopyDone package(keep alive now not necessary we preparing to complete). WalSndWaitForWal after get CopyDone exit from loop. With apply this patch I get next measurements