anchor.diff:  -export([init/1, code_change/3, terminate/2]).
   
   %% state callbacks
  --export([auth/2, initializing/2, on_message/2]).
  +-export([auth/2, initializing/2, on_message/2, on_replication/2]).
   
   -include("epgsql.hrl").
   -include("epgsql_binary.hrl").
  
positive.code:     notify_async(State, {notification, Channel1, Pid, Payload1}),
      {noreply, State};
  
  %% CopyBothResponse
  on_message({?COPY_BOTH_RESPONSE, _Data}, State) ->
      State2 = finish(State, ok),
      {noreply, State2};
  
  %% CopyData for COPY command. COPY command not supported yet.
  on_message({?COPY_DATA, _Data}, #state{repl_cbmodule = undefined, repl_receiver = undefined} = State) ->
      {stop, {error, copy_command_not_supported}, State};
  
  %% CopyData for Replication mode
  on_message({?COPY_DATA, <<?PRIMARY_KEEPALIVE_MESSAGE:8, LSN:?int64, _Timestamp:?int64, ReplyRequired:8>>},
      #state{repl_last_flushed_lsn = LastFlushedLSN, repl_last_applied_lsn = LastAppliedLSN} = State) ->
      case ReplyRequired of
          1 ->
              send(State, ?COPY_DATA,
                  epgsql_wire:encode_standby_status_update(LSN, LastFlushedLSN, LastAppliedLSN)),
              {noreply, State#state{repl_feedback_required = false, repl_last_received_lsn = LSN}};
          _ ->
              {noreply, State#state{repl_feedback_required = true, repl_last_received_lsn = LSN}}
      end;
  
  %% CopyData for Replication mode. with async messages
  on_message({?COPY_DATA, <<?X_LOG_DATA, StartLSN:?int64, EndLSN:?int64, _Timestamp:?int64, WALRecord/binary>>},
      #state{repl_cbmodule = undefined, repl_receiver = Receiver} = State) ->
      Receiver ! {epgsql, self(), {x_log_data, StartLSN, EndLSN, WALRecord}},
      {noreply, State#state{repl_feedback_required = true, repl_last_received_lsn = EndLSN}};
  
  %% CopyData for Replication mode. with callback method
  on_message({?COPY_DATA, <<?X_LOG_DATA, StartLSN:?int64, EndLSN:?int64, _Timestamp:?int64, WALRecord/binary>>},
      #state{repl_cbmodule = CbModule, repl_cbstate = CbState, repl_receiver = undefined} = State) ->
      {ok, LastFlushedLSN, LastAppliedLSN, NewCbState} =
          CbModule:handle_x_log_data(StartLSN, EndLSN, WALRecord, CbState),
      {noreply, State#state{repl_feedback_required = true, repl_last_received_lsn = EndLSN,
          repl_last_flushed_lsn = LastFlushedLSN, repl_last_applied_lsn = LastAppliedLSN,
          repl_cbstate = NewCbState}}.
  
negative.code:             PluginOpts -> [Sql1 , " (", PluginOpts, ")"]
          end,
  
      Repl2 =
          case Callback of
              Pid when is_pid(Pid) -> Repl#repl{receiver = Pid};
              Module -> Repl#repl{cbmodule = Module, cbstate = CbInitState}
          end,
  
      Hex = [H || H <- WALPosition, H =/= $/],
      {ok, [LSN], _} = io_lib:fread("~16u", Hex),
  
      State2 = State#state{repl = Repl2#repl{last_flushed_lsn = LSN,
                                             last_applied_lsn = LSN},
                           handler = on_replication},
  
      send(State2, ?SIMPLEQUERY, [Sql2, 0]),
      {noreply, State2}.
  
  start_ssl(S, Flag, Opts, State) ->
      ok = gen_tcp:send(S, <<8:?int32, 80877103:?int32>>),
  