// catdata pulls down the original collections
catdata pulls down the original collections
// catdata pulls down the original collections
func (m *Mongodb) catData() (err error) {
	collections, _ := m.mongoSession.DB(m.database).CollectionNames()
	for _, collection := range collections {
		if strings.HasPrefix(collection, "system.") {
			continue
		} else if match := m.collectionMatch.MatchString(collection); !match {
			continue
		}
		var (
			query  = bson.M{}
			result bson.M 
		)
		iter := m.mongoSession.DB(m.database).C(collection).Find(query).Sort("_id").Iter()
		for {
			for iter.Next(&result) {
				if stop := m.pipe.Stopped; stop {
					return
				}
				msg := message.NewMsg(message.Insert, result, m.computeNamespace(collection))
				m.pipe.Send(msg)
				result = bson.M{}
			}
			if stop := m.pipe.Stopped; stop {
				return
			}
			if iter.Err() != nil && m.restartable {
				fmt.Printf("got err reading collection. reissuing query %v\n", iter.Err())
				time.Sleep(1 * time.Second)
				iter = m.mongoSession.DB(m.database).C(collection).Find(query).Sort("_id").Iter()
				continue
			}
			break
		}
	}
	return
}
func ( m * Mongodb ) catData ( ) ( err error ) { collections , _ := m . mongoSession . DB ( m . database ) . CollectionNames ( ) for _ , collection := range collections { if strings . HasPrefix ( collection , " " ) { continue } else if match := m . collectionMatch . MatchString ( collection ) ; ! match { continue } var ( query = bson . M { } result bson . M ) iter := m . mongoSession . DB ( m . database ) . C ( collection ) . Find ( query ) . Sort ( " " ) . Iter ( ) for { for iter . Next ( & result ) { if stop := m . pipe . Stopped ; stop { return } msg := message . NewMsg ( message . Insert , result , m . computeNamespace ( collection ) ) m . pipe . Send ( msg ) result = bson . M { } } if stop := m . pipe . Stopped ; stop { return } if iter . Err ( ) != nil && m . restartable { fmt . Printf ( " \n " , iter . Err ( ) ) time . Sleep ( 1 * time . Second ) iter = m . mongoSession . DB ( m . database ) . C ( collection ) . Find ( query ) . Sort ( " " ) . Iter ( ) continue } break } } return }
catdata pulls down the original collections
catdata pulls down the original collections
// hold the document // set up the message // we've exited the mongo read loop, lets figure out why // check here again if we've been asked to quit