message Entry { //Entryは単一のサブスクリプションイベント構造です。1つのイベントはMySQLの1つのbinlog eventに相当します。
 Header header = 1;  //イベントヘッダー
 Event event   = 2;  //イベントボディ
}


message Header {
 int32       version        = 1;   //Entryプロトコルバージョン
 SourceType  sourceType     = 2;   //ソースデータベースのタイプ情報。MySQL、Oracleなどのタイプが含まれます。
 MessageType messageType    = 3;   //メッセージのタイプ、つまりEventのタイプです。BEGIN、COMMIT、DMLなどが含まれます。
 uint32 timestamp           = 4;   //Eventの元のbinlogにおけるタイムスタンプ
 int64  serverId            = 5;   //ソースのserverId
 string fileName            = 6;   //ソースbinlogのファイル名
 uint64 position            = 7;   //ソースbinlogファイル内のイベントのオフセット
 string gtid                = 8;   //現在のトランザクションのgtid
 string schemaName          = 9;   //変更の影響を受けるスキーマ
 string tableName           = 10;  //変更の影響を受けるテーブル
 uint64 seqId               = 11;  //グローバル増分シーケンス番号
 uint64 eventIndex          = 12;  //大きなeventをシャード化する場合、各シャードは0から番号が振られます。現在のバージョンでは意味がなく、今後の拡張用に予約されています。
 bool   isLast              = 13;  //現在のeventがeventシャードの最後のブロックであるかどうか。最後のブロックである場合はtrueです。現在のバージョンでは意味がなく、今後の拡張用に予約されています。
 repeated KVPair properties = 15;
}


message Event {
 BeginEvent      beginEvent      = 1;  //binlog内のbeginイベント
 DMLEvent        dmlEvent        = 2;  //binlog内のdmlイベント
 CommitEvent     commitEvent     = 3;  //binlog内のcommitイベント
 DDLEvent        ddlEvent        = 4;  //binlog内のddlイベント
 RollbackEvent   rollbackEvent   = 5;  //rollbackイベント。現在のバージョンでは意味がありません。
 HeartbeatEvent  heartbeatEvent  = 6;  //ソースデータベースが定期的に送信するハートビートイベント
 CheckpointEvent checkpointEvent = 7;  //購読バックエンドが追加するcheckpointイベントです。10秒ごとに1つ自動生成され、Kafkaの生産および消費場所の管理に使用されます。
 repeated KVPair properties      = 15;
}

message Entries {
     repeated Entry items = 1; //entry list
}

message Envelope {
     int32  version                  = 1; //プロトコルバージョン。data内容のデコード方法を決定します。
     uint32 total                    = 2;
     uint32 index                    = 3;
     bytes  data                     = 4; //現在のversionが1の場合、data内のデータはEntriesがPBシリアル化された結果であることを示します。
     repeated KVPair properties      = 15;
}

 // Kafka メッセージの Value 値を Envelope に変換します。
 envelope := subscribe.Envelope{}
 err := proto.Unmarshal(msg.Value, &envelope)

 if envelope.Index == 0 {
     pmc.completeMsg = envelope
 } else {
     // 分割された Entries のバイナリシーケンスを結合します。
     pmc.completeMsg.Data = append(pmc.completeMsg.Data, envelope.Data...)
 }
 if envelope.Index < envelope.Total-1 {
     return nil
 }
 // Envelope.Data を Entries にデシリアライズします。
 entries := subscribe.Entries{}
 err = proto.Unmarshal(pmc.completeMsg.Data, &entries)