Combine syncer localReader into one interface #215

3pointer · 2019-07-25T11:03:31Z

What problem does this PR solve?

Combine two Syncer members (Syncer.syncer and Syncer.localReader) into one variable (streamerProducer), base on two reasons:

Obviously Syncer.syncer and Syncer.localReader has same ability to generate stream which can generate events, And Syncer only need one not two at any time
And after put them in one interface, it's easy for any implementations who implement StreamerProducer and Stream interfaces to take place of Syncer.streamerProducer, such as a mock test

What is changed and how it works?

Combine two Syncer members into one streamerProducer

Check List

Tests

Unit test
Integration test

3pointer · 2019-07-25T11:06:39Z

pkg/streamer/reader.go

@@ -383,12 +383,11 @@ func (r *BinlogReader) updateUUIDs() error {
 }

 // Close closes BinlogReader.
-func (r *BinlogReader) Close() error {
+func (r *BinlogReader) Close() {


this implementation aligned with *BinlogSyncer Close() in go-mysql

codecov · 2019-07-25T11:09:55Z

Codecov Report

Merging #215 into master will decrease coverage by 0.5883%.
The diff coverage is 53.125%.

@@               Coverage Diff                @@
##             master       #215        +/-   ##
================================================
- Coverage   59.4612%   58.8729%   -0.5884%     
================================================
  Files           123        123                
  Lines         14887      14054       -833     
================================================
- Hits           8852       8274       -578     
+ Misses         5150       4953       -197     
+ Partials        885        827        -58

amyangfei · 2019-07-25T15:16:40Z

syncer/syncer.go

-	if s.localReader != nil {
-		s.localReader.Close()
+	if s.streamerProducer != nil {
+		rr, ok := s.streamerProducer.(*remoteBinlogReader)


switch r := s.streamerProducer.(type) { case *remoteBinlogReader: // process remote binlog reader case *localBinlogReader: // process local binlog reader }

use type-switch is more clear

amyangfei · 2019-07-25T15:37:26Z

syncer/syncer.go

-	} else if s.binlogType == LocalBinlog {
-		s.streamer, err = s.getBinlogStreamer(s.localReader, pos)
-	}
+	s.streamer, err = s.streamerProducer.generateStreamer(pos)


we can remove getBinlogStreamer, getLocalBinlogStreamer and getRemoteBinlogStreamer after we extract the interface.

amyangfei · 2019-07-25T15:38:57Z

syncer/syncer.go

@@ -79,6 +79,46 @@ const (
 	LocalBinlog
 )

+// StreamerProducer provide the ability to generate binlog streamer by StartSync()


Suggested change

// StreamerProducer provide the ability to generate binlog streamer by StartSync()

// StreamerProducer provides the ability to generate binlog streamer by StartSync()

amyangfei · 2019-07-25T15:39:24Z

syncer/syncer.go

@@ -79,6 +79,46 @@ const (
 	LocalBinlog
 )

+// StreamerProducer provide the ability to generate binlog streamer by StartSync()
+// but go-mysql StartSync() return (struct, err) rather than (interface, err)


Suggested change

// but go-mysql StartSync() return (struct, err) rather than (interface, err)

// but go-mysql StartSync() returns (struct, err) rather than (interface, err)

amyangfei · 2019-07-25T15:40:49Z

syncer/syncer.go

+	}()
+	if r.EnableGTID {
+		// NOTE: our (per-table based) checkpoint does not support GTID yet
+		return nil, errors.New("[syncer] now support GTID mode yet")


Suggested change

return nil, errors.New("[syncer] now support GTID mode yet")

return nil, errors.New("[syncer] not support GTID mode yet")

amyangfei · 2019-07-25T15:46:33Z

pkg/streamer/reader_test.go

@@ -589,7 +589,6 @@ func (t *testReaderSuite) TestStartSync(c *C) {
 	// NOTE: load new UUIDs dynamically not supported yet

 	// close the reader
-	err = r.Close()


Although Close doesn't return error anymore, we'd better reserve it to form a complete usage flow.

Yes, I forgot to use it, it's been fixed already in the latest commit

IANTHEREAL · 2019-07-26T11:27:27Z

pkg/streamer/reader_test.go

@@ -589,8 +589,8 @@ func (t *testReaderSuite) TestStartSync(c *C) {
 	// NOTE: load new UUIDs dynamically not supported yet

 	// close the reader
-	err = r.Close()
 	c.Assert(err, IsNil)


already added back

amyangfei

LGTM

IANTHEREAL · 2019-07-29T04:22:11Z

syncer/syncer.go

-	if s.syncer != nil {
-		err := s.closeBinlogSyncer(s.syncer)
-		s.syncer = nil
+	if s.streamerProducer != nil {


should we check whether it's remote binlog reader?

only remote binlog reader call this function currently, but adding a check is better

IANTHEREAL · 2019-07-29T04:24:16Z

Rest LGTM

IANTHEREAL · 2019-07-29T05:06:27Z

syncer/syncer.go

+				return nil, errors.Trace(err)
+			}
+		case *localBinlogReader:
+			return nil, errors.New("[syncer] not support local relay reader reopen currently")


Suggested change

return nil, errors.New("[syncer] not support local relay reader reopen currently")

return nil, errors.New("don't support reopen %T currently", r)

IANTHEREAL · 2019-07-29T05:07:30Z

syncer/syncer.go

+	}()
+	if r.EnableGTID {
+		// NOTE: our (per-table based) checkpoint does not support GTID yet
+		return nil, errors.New("[syncer] not support GTID mode yet")


Suggested change

return nil, errors.New("[syncer] not support GTID mode yet")

return nil, errors.New("don't support open streamer with GTID mode")

IANTHEREAL · 2019-07-29T05:09:33Z

syncer/syncer.go

-		if err != nil {
-			return nil, errors.Trace(err)
+	if s.streamerProducer != nil {
+		switch s.streamerProducer.(type) {


Suggested change

switch s.streamerProducer.(type) {

switch r := s.streamerProducer.(type) {

IANTHEREAL · 2019-07-29T05:09:53Z

syncer/syncer.go

+	if s.streamerProducer != nil {
+		switch s.streamerProducer.(type) {
+		case *remoteBinlogReader:
+			err := s.closeBinlogSyncer(s.streamerProducer.(*remoteBinlogReader).reader)


Suggested change

err := s.closeBinlogSyncer(s.streamerProducer.(*remoteBinlogReader).reader)

err := s.closeBinlogSyncer(r.reader)

IANTHEREAL · 2019-07-29T05:10:20Z

syncer/syncer.go

+			if err != nil {
+				return nil, errors.Trace(err)
+			}
+		case *localBinlogReader:


Suggested change

case *localBinlogReader:

default:

IANTHEREAL · 2019-07-29T05:20:26Z

/run-all-tests

3pointer · 2019-07-29T05:35:18Z

/run-all-tests

IANTHEREAL

Good Job! LGTM

* extract syncer localReader to one interface * use generateStreamer wrap function * update some implementations and remove useless code * remove useless err check * update check syncer reopen * update error message

3pointer added 2 commits July 25, 2019 16:50

extract syncer localReader to one interface

e3fb0c2

use generateStreamer wrap function

c98b224

3pointer commented Jul 25, 2019

View reviewed changes

3pointer changed the title ~~extract syncer localReader to one interface~~ Combine syncer localReader into one interface Jul 25, 2019

3pointer added type/feature-request This issue is a feature request priority/normal Minor change, requires approval from ≥1 primary reviewer status/PTAL This PR is ready for review. Add this label back after committing new changes type/feature New feature labels Jul 25, 2019

amyangfei reviewed Jul 25, 2019

View reviewed changes

update some implementations and remove useless code

a523b0a

IANTHEREAL reviewed Jul 26, 2019

View reviewed changes

remove useless err check

4654ad8

amyangfei reviewed Jul 29, 2019

View reviewed changes

amyangfei added status/LGT1 One reviewer already commented LGTM and removed status/PTAL This PR is ready for review. Add this label back after committing new changes labels Jul 29, 2019

IANTHEREAL reviewed Jul 29, 2019

View reviewed changes

update check syncer reopen

fde4349

IANTHEREAL reviewed Jul 29, 2019

View reviewed changes

3pointer force-pushed the update_syncer branch from 7534f21 to 0c9d63f Compare July 29, 2019 05:12

update error message

22af3c7

3pointer force-pushed the update_syncer branch from 0c9d63f to 22af3c7 Compare July 29, 2019 05:17

IANTHEREAL approved these changes Jul 29, 2019

View reviewed changes

IANTHEREAL added status/LGT2 Two reviewers already commented LGTM, ready for merge and removed status/LGT1 One reviewer already commented LGTM labels Jul 29, 2019

3pointer merged commit 4abf188 into pingcap:master Jul 29, 2019

amyangfei mentioned this pull request Aug 1, 2019

increase DM unit test/integration test code coverage #70

Open

22 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Combine syncer localReader into one interface #215

Combine syncer localReader into one interface #215

3pointer commented Jul 25, 2019

3pointer Jul 25, 2019

codecov bot commented Jul 25, 2019 •

edited

Loading

amyangfei Jul 25, 2019

3pointer Jul 26, 2019

amyangfei Jul 25, 2019

3pointer Jul 26, 2019

amyangfei Jul 25, 2019

3pointer Jul 26, 2019

amyangfei Jul 25, 2019

3pointer Jul 26, 2019

amyangfei Jul 25, 2019

3pointer Jul 26, 2019

amyangfei Jul 25, 2019

3pointer Jul 26, 2019

IANTHEREAL Jul 26, 2019

3pointer Jul 26, 2019

amyangfei left a comment

IANTHEREAL Jul 29, 2019

3pointer Jul 29, 2019

IANTHEREAL commented Jul 29, 2019

IANTHEREAL Jul 29, 2019 •

edited

Loading

IANTHEREAL Jul 29, 2019

IANTHEREAL Jul 29, 2019

IANTHEREAL Jul 29, 2019

IANTHEREAL Jul 29, 2019

IANTHEREAL commented Jul 29, 2019

3pointer commented Jul 29, 2019

IANTHEREAL left a comment

	// StreamerProducer provide the ability to generate binlog streamer by StartSync()
	// StreamerProducer provides the ability to generate binlog streamer by StartSync()

	// but go-mysql StartSync() return (struct, err) rather than (interface, err)
	// but go-mysql StartSync() returns (struct, err) rather than (interface, err)

	return nil, errors.New("[syncer] now support GTID mode yet")
	return nil, errors.New("[syncer] not support GTID mode yet")

	return nil, errors.New("[syncer] not support local relay reader reopen currently")
	return nil, errors.New("don't support reopen %T currently", r)

	return nil, errors.New("[syncer] not support GTID mode yet")
	return nil, errors.New("don't support open streamer with GTID mode")

	switch s.streamerProducer.(type) {
	switch r := s.streamerProducer.(type) {

	err := s.closeBinlogSyncer(s.streamerProducer.(*remoteBinlogReader).reader)
	err := s.closeBinlogSyncer(r.reader)

Combine syncer localReader into one interface #215

Combine syncer localReader into one interface #215

Conversation

3pointer commented Jul 25, 2019

What problem does this PR solve?

What is changed and how it works?

Check List

Choose a reason for hiding this comment

codecov bot commented Jul 25, 2019 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amyangfei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

IANTHEREAL commented Jul 29, 2019

IANTHEREAL Jul 29, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

IANTHEREAL commented Jul 29, 2019

3pointer commented Jul 29, 2019

IANTHEREAL left a comment

Choose a reason for hiding this comment

codecov bot commented Jul 25, 2019 •

edited

Loading

IANTHEREAL Jul 29, 2019 •

edited

Loading