processMessage processes a logical replication message as appropriate. A couple important aspects: 1. Relation messages describe tables being replicated and are used to build a type map for decoding tuples 2. INSERT/UPDATE/DELETE messages describe changes to rows that must be applied to the replica.
( xld pglogrepl.XLogData, state *replicationState, )
| 549 | // |
| 550 | // Returns a boolean true if the message was a commit that should be acknowledged, and an error if one occurred. |
| 551 | func (r *LogicalReplicator) processMessage( |
| 552 | xld pglogrepl.XLogData, |
| 553 | state *replicationState, |
| 554 | ) (bool, error) { |
| 555 | walData := xld.WALData |
| 556 | logicalMsg, err := pglogrepl.ParseV2(walData, state.inStream) |
| 557 | if err != nil { |
| 558 | return false, err |
| 559 | } |
| 560 | |
| 561 | log.Printf("XLogData (%T) => WALStart %s ServerWALEnd %s ServerTime %s", logicalMsg, xld.WALStart, xld.ServerWALEnd, xld.ServerTime) |
| 562 | state.lastReceivedLSN = xld.ServerWALEnd |
| 563 | |
| 564 | switch logicalMsg := logicalMsg.(type) { |
| 565 | case *pglogrepl.RelationMessageV2: |
| 566 | state.relations[logicalMsg.RelationID] = logicalMsg |
| 567 | case *pglogrepl.BeginMessage: |
| 568 | // Indicates the beginning of a group of changes in a transaction. |
| 569 | // This is only sent for committed transactions. We won't get any events from rolled back transactions. |
| 570 | |
| 571 | if state.lastWrittenLSN > logicalMsg.FinalLSN { |
| 572 | log.Printf("Received stale message, ignoring. Last written LSN: %s Message LSN: %s", state.lastWrittenLSN, logicalMsg.FinalLSN) |
| 573 | state.processMessages = false |
| 574 | return false, nil |
| 575 | } |
| 576 | |
| 577 | state.processMessages = true |
| 578 | state.currentTransactionLSN = logicalMsg.FinalLSN |
| 579 | |
| 580 | log.Printf("BeginMessage: %v", logicalMsg) |
| 581 | err = r.replicateQuery(state.replicaConn, "START TRANSACTION") |
| 582 | if err != nil { |
| 583 | return false, err |
| 584 | } |
| 585 | case *pglogrepl.CommitMessage: |
| 586 | log.Printf("CommitMessage: %v", logicalMsg) |
| 587 | err = r.replicateQuery(state.replicaConn, "COMMIT") |
| 588 | if err != nil { |
| 589 | return false, err |
| 590 | } |
| 591 | state.processMessages = false |
| 592 | |
| 593 | return true, nil |
| 594 | case *pglogrepl.InsertMessageV2: |
| 595 | if !state.processMessages { |
| 596 | log.Printf("Received stale message, ignoring. Last written LSN: %s Message LSN: %s", state.lastWrittenLSN, xld.ServerWALEnd) |
| 597 | return false, nil |
| 598 | } |
| 599 | |
| 600 | rel, ok := state.relations[logicalMsg.RelationID] |
| 601 | if !ok { |
| 602 | log.Fatalf("unknown relation ID %d", logicalMsg.RelationID) |
| 603 | } |
| 604 | |
| 605 | columnStr := strings.Builder{} |
| 606 | valuesStr := strings.Builder{} |
| 607 | for idx, col := range logicalMsg.Tuple.Columns { |
| 608 | if idx > 0 { |
no test coverage detected