Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

added query event in dummy event handler #791

Open
wants to merge 4 commits into
base: master
Choose a base branch
from

Conversation

axfor
Copy link
Contributor

@axfor axfor commented May 8, 2023

added more query event

  • onQueryEvent query event include(create user,drop user,create index event,etd.)

canal/handler.go Outdated Show resolved Hide resolved
canal/handler.go Outdated Show resolved Hide resolved
canal/handler.go Outdated Show resolved Hide resolved
@axfor axfor changed the title added onCreateUser and onDropUser and OnGrant and onQueryEvent added query event in dummy event handler Jun 13, 2023
@axfor axfor closed this Jun 13, 2023
@axfor axfor reopened this Jun 13, 2023
canal/handler.go Outdated Show resolved Hide resolved
canal/sync.go Outdated Show resolved Hide resolved
Signed-off-by: axfor <aixiaoxiang2009@hotmail.com>
@axfor axfor force-pushed the feature-axx branch 2 times, most recently from 52ce4ff to d6fe43b Compare June 23, 2023 11:29
Signed-off-by: axfor <aixiaoxiang2009@hotmail.com>
canal/handler.go Outdated Show resolved Hide resolved
canal/handler.go Outdated Show resolved Hide resolved
canal/sync.go Outdated
@@ -336,3 +341,37 @@ func (c *Canal) CatchMasterPos(timeout time.Duration) error {

return c.WaitUntilPos(pos, timeout)
}

type Position struct {
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why do we need it?

Copy link
Contributor Author

@axfor axfor Jun 24, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Based on mysql.Position adds SavePos and Force fields

I think queryEvent contains multiple sub-events, pos save logic, and it is up to the end-user event to decide whether to save or not


image
func (h *DummyEventHandler) OnQueryEvent(header *replication.EventHeader, stmt ast.StmtNode, pos *Position, e *replication.QueryEvent) error { 
         //....
        pos.SavePos =true
	return nil
}

Of course, it can also be handled uniformly in the main loop of canal/sync.go events.

https://github.com/axfor/go-mysql/blob/feature-axx/canal/sync.go#L161

func (c *Canal) startSyncer() (*replication.BinlogStreamer, error) { 
        //....
       for _, stmt := range stmts {
		nodes := parseDDLStmt(stmt)
		if len(nodes) > 0 {
		      // OnDDL
		} else {
			err := c.handleQueryEvent(ev.Header, stmt, posInfo, e)
			if err != nil {
				c.cfg.Logger.Errorf("handle query event err %v", err)
			} else {
				savePos = true   // for save pos
				force = true   // for save pos
			}
               }
              //....
       } 
      //....
}

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it is up to the end-user event to decide whether to save or not

I think it's more obvious to use return value to do it.

OnQueryEvent(ev *replication.BinlogEvent, e *replication.QueryEvent, stmt ast.StmtNode, pos mysql.Position) (savePos bool, force bool, err error)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I'll modify it

axfor and others added 2 commits June 24, 2023 11:37
update commit of onQueryEvent

Co-authored-by: lance6716 <lance6716@gmail.com>
Signed-off-by: axfor <aixiaoxiang2009@hotmail.com>
@@ -17,18 +18,22 @@ type EventHandler interface {
OnGTID(header *replication.EventHeader, gtid mysql.GTIDSet) error
// OnPosSynced Use your own way to sync position. When force is true, sync position immediately.
OnPosSynced(header *replication.EventHeader, pos mysql.Position, set mysql.GTIDSet, force bool) error
// OnQueryEvent is query event include (create user, drop user, create index, etc.)
// Note: the OnQueryEvent has lower priority than OnDDL even
OnQueryEvent(header *replication.EventHeader, stmt ast.StmtNode, pos mysql.Position, e *replication.QueryEvent) (bool, bool, error)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please use named return in this interface, or explain what's the meaning of two bools in comments. So users of this library will understand when they implement this interface.

canal/sync.go Show resolved Hide resolved
Comment on lines +361 to +364
if err != nil {
return savePos, force, errors.Trace(err)
}
return savePos, force, nil
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
if err != nil {
return savePos, force, errors.Trace(err)
}
return savePos, force, nil
return savePos, force, errors.Trace(err)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

None yet

3 participants