// Start the adaptor as a source
Start the adaptor as a source
// Start the adaptor as a source
func (r *Rethinkdb) Start() error {
	tables, err := gorethink.DB(r.database).TableList().Run(r.client)
	if err != nil {
		return err
	}
	defer tables.Close()
	var (
		wg    sync.WaitGroup
		table string
	)
	for tables.Next(&table) {
		if match := r.tableMatch.MatchString(table); !match {
			if r.debug {
				fmt.Printf("table, %s, didn't match\n", table)
			}
			continue
		}
		wg.Add(1)
		start := make(chan bool)
		if r.tail {
			go r.startupChangesForTable(table, start, &wg)
		}
		go func(table string) {
			defer wg.Done()
			if err := r.sendAllDocuments(table); err != nil {
				r.pipe.Err <- err
			} else if r.tail {
				start <- true
			}
			close(start)
		}(table)
	}
	wg.Wait()
	return nil
}
func ( r * Rethinkdb ) Start ( ) error { tables , err := gorethink . DB ( r . database ) . TableList ( ) . Run ( r . client ) if err != nil { return err } defer tables . Close ( ) var ( wg sync . WaitGroup table string ) for tables . Next ( & table ) { if match := r . tableMatch . MatchString ( table ) ; ! match { if r . debug { fmt . Printf ( " \n " , table ) } continue } wg . Add ( 1 ) start := make ( chan bool ) if r . tail { go r . startupChangesForTable ( table , start , & wg ) } go func ( table string ) { defer wg . Done ( ) if err := r . sendAllDocuments ( table ) ; err != nil { r . pipe . Err <- err } else if r . tail { start <- true } close ( start ) } ( table ) } wg . Wait ( ) return nil }
Start the adaptor as a source
Start the adaptor as a source