null if the message isn't supported
* in that protocol version.
*/
public abstract byte[] getBytes(short protocolVersion);
/**
* Whether this message has an encoding for a peer which negotiated the
* provided replication protocol version.
*
* A message which has none is dropped by {@link Session#publish(ReplicationMsg)}
* - its {@link #getBytes(short)} returns null - with nothing on
* the wire and nothing in the log, so a caller which goes on to report what
* the peer was told must ask this before it publishes.
*
* @param protocolVersion
* The protocol version negotiated with the peer.
* @return true if this message can be encoded for that version,
* false if it has no encoding for it at all.
*/
public boolean isEncodableFor(short protocolVersion)
{
return true;
}
/**
* Generates a ReplicationMsg from its encoded form. This un-serialization is
* done taking into account the various supported replication protocol
* versions.
*
* @param buffer
* The encode form of the ReplicationMsg.
* @param protocolVersion
* The version to use to decode the msg.
* @return The generated SynchronizationMessage.
* @throws DataFormatException
* If the encoded form was not a valid msg.
* @throws NotSupportedOldVersionPDUException
* If the PDU is part of an old protocol version and we do not
* support it.
*/
public static ReplicationMsg generateMsg(byte[] buffer, short protocolVersion)
throws DataFormatException, NotSupportedOldVersionPDUException
{
switch (buffer[0])
{
case MSG_TYPE_SERVER_START_V1:
throw new NotSupportedOldVersionPDUException("Server Start",
ProtocolVersion.REPLICATION_PROTOCOL_V1, buffer[0]);
case MSG_TYPE_REPL_SERVER_INFO_V1:
throw new NotSupportedOldVersionPDUException("Replication Server Info",
ProtocolVersion.REPLICATION_PROTOCOL_V1, buffer[0]);
case MSG_TYPE_MODIFY:
return new ModifyMsg(buffer);
case MSG_TYPE_MODIFY_V1:
return ModifyMsg.createV1(buffer);
case MSG_TYPE_ADD:
case MSG_TYPE_ADD_V1:
return new AddMsg(buffer);
case MSG_TYPE_DELETE:
case MSG_TYPE_DELETE_V1:
return new DeleteMsg(buffer);
case MSG_TYPE_MODIFYDN:
case MSG_TYPE_MODIFYDN_V1:
return new ModifyDNMsg(buffer);
case MSG_TYPE_ACK:
return new AckMsg(buffer);
case MSG_TYPE_SERVER_START:
return new ServerStartMsg(buffer);
case MSG_TYPE_REPL_SERVER_START:
case MSG_TYPE_REPL_SERVER_START_V1:
return new ReplServerStartMsg(buffer);
case MSG_TYPE_WINDOW:
return new WindowMsg(buffer);
case MSG_TYPE_HEARTBEAT:
return new HeartbeatMsg(buffer);
case MSG_TYPE_INITIALIZE_REQUEST:
return new InitializeRequestMsg(buffer, protocolVersion);
case MSG_TYPE_INITIALIZE_TARGET:
return new InitializeTargetMsg(buffer, protocolVersion);
case MSG_TYPE_ENTRY:
return new EntryMsg(buffer, protocolVersion);
case MSG_TYPE_DONE:
return new DoneMsg(buffer);
case MSG_TYPE_ERROR:
return new ErrorMsg(buffer, protocolVersion);
case MSG_TYPE_RESET_GENERATION_ID:
return new ResetGenerationIdMsg(buffer);
case MSG_TYPE_WINDOW_PROBE:
return new WindowProbeMsg(buffer);
case MSG_TYPE_TOPOLOGY:
return new TopologyMsg(buffer, protocolVersion);
case MSG_TYPE_REPL_SERVER_MONITOR_REQUEST:
return new MonitorRequestMsg(buffer);
case MSG_TYPE_REPL_SERVER_MONITOR:
return new MonitorMsg(buffer, protocolVersion);
case MSG_TYPE_START_SESSION:
return new StartSessionMsg(buffer, protocolVersion);
case MSG_TYPE_CHANGE_STATUS:
return new ChangeStatusMsg(buffer);
case MSG_TYPE_GENERIC_UPDATE:
return new UpdateMsg(buffer);
case MSG_TYPE_START_ECL:
case MSG_TYPE_START_ECL_SESSION:
case MSG_TYPE_ECL_UPDATE:
// Legacy versions never sent such messages to other instances (other JVMs).
// They were only used in the combined DS-RS case.
// It is safe to totally ignore these values since code now uses the ChangelogBackend.
return null;
case MSG_TYPE_CT_HEARTBEAT:
return new ChangeTimeHeartbeatMsg(buffer, protocolVersion);
case MSG_TYPE_REPL_SERVER_START_DS:
return new ReplServerStartDSMsg(buffer);
case MSG_TYPE_STOP:
return new StopMsg(buffer);
case MSG_TYPE_INITIALIZE_RCV_ACK:
return new InitializeRcvAckMsg(buffer);
case MSG_TYPE_REPLICA_OFFLINE:
return new ReplicaOfflineMsg(buffer);
default:
throw new DataFormatException("received message with unknown type");
}
}
}