func changeFeedHelper(cursor *r.Cursor, changeEventName string,
	send chan<- Message, stop <-chan bool) {
	change := make(chan r.ChangeResponse)
	cursor.Listen(change)
	for {
		eventName := ""
		var data interface{}
		select {
		case <-stop:
			cursor.Close()
			return
		case val := <-change:
			if val.NewValue != nil && val.OldValue == nil {
				eventName = changeEventName + " add"
				data = val.NewValue
			} else if val.NewValue == nil && val.OldValue != nil {
				eventName = changeEventName + " remove"
				data = val.OldValue
			} else if val.NewValue != nil && val.OldValue != nil {
				eventName = changeEventName + " edit"
				data = val.NewValue
			}
			send <- Message{eventName, data}
		}
	}
}

func createSubscription(cursor *r.Cursor, client *Client, entityName string, stopKey int) {
	recordChan := make(chan r.ChangeResponse)
	cursor.Listen(recordChan)

	stopChan := client.NewStopChannel(stopKey)

	go func() {
		for {
			select {
			case <-stopChan:
				cursor.Close()
				return
			case change := <-recordChan:
				var command string
				var data interface{}
				if change.NewValue != nil && change.OldValue == nil {
					command = entityName + " add"
					data = change.NewValue
				}
				if change.NewValue == nil && change.OldValue != nil {
					command = entityName + " remove"
					data = change.OldValue
					fmt.Println("sent '" + entityName + " remove' msg")
				}
				if change.NewValue != nil && change.OldValue != nil {
					command = entityName + " edit"
					data = change.NewValue
				}
				client.send <- SocketMessage{command, data}
				fmt.Println("sent '"+command+"' msg ", data)
			}
		}
	}()
}

func printResult(result *r.Cursor) {
	var kv map[string]interface{}
	i := 0
	for result.Next(&kv) {
		fmt.Println("result", i)
		fmt.Println(kv)
		i = i + 1
	}
}

func processMultipleCursor(
	cursor *gorethink.Cursor,
	messageConstructor func() proto.Message,
) ([]interface{}, error) {
	var data []string
	if err := cursor.All(&data); err != nil {
		return nil, err
	}
	result := make([]interface{}, len(data))
	for i, datum := range data {
		message := messageConstructor()
		if err := jsonpb.UnmarshalString(datum, message); err != nil {
			return nil, err
		}
		result[i] = message
	}
	return result, nil
}

func (d *dbService) GetUser(email string, isApp bool, appId string) (user models.JSON, err error) {
	var c *r.Cursor
	if isApp {
		c, err = d.Run(
			d.Db().Table(APPS_USERS_TABLE).GetAllByIndex(APPS_USERS_TABLE_EMAILAPPID_INDEX, []interface{}{email, appId}).Nth(0),
		)
	} else {
		c, err = d.Run(
			d.Db().Table(USERS_TABLE).GetAllByIndex(USERS_TABLE_EMAIL_INDEX, email).Nth(0),
		)
	}

	if err != nil {
		return
	}

	err = c.One(&user)
	return
}

func InitDB(dbName string) (session *r.Session, err error) {

	var cursor *r.Cursor

	session, err = r.Connect(r.ConnectOpts{
		Address:  os.Getenv("RETHINKDB_URL"),
		Database: dbName,
		MaxIdle:  10,
		Timeout:  time.Second * 10,
	})

	if err != nil {
		l.Println("Connect", err)
		return
	}

	cursor, err = r.DBList().Contains(dbName).Do(func(row r.Term) r.Term {
		return r.Branch(
			row.Eq(true),
			nil,
			r.DBCreate(dbName),
		)
	}).Run(session)
	defer cursor.Close()

	for _, table := range tables {

		cursor, err = r.DB(dbName).TableList().Contains(table.name).Do(func(row r.Term) r.Term {
			return r.Branch(
				row.Eq(true),
				nil,
				r.DB(dbName).TableCreate(table.name, table.options),
			)
		}).Run(session)
		defer cursor.Close()

		if err != nil {
			return
		}
	}

	return
}

func packageCursorToUrl(c *r.Cursor) (UrlMap, error) {
	rows := []UrlMap{}
	err := c.All(&rows)
	return rows[0], err
}

func (bmo *BMO) Compute(input *os.File) {

	var err error
	var cur *r.Cursor
	var session *r.Session

	// set up database connection pool
	session, err = r.Connect(r.ConnectOpts{
		Addresses:     bmo.nodes,
		Database:      bmo.database,
		DiscoverHosts: true,
	})
	session.SetMaxOpenConns(POOL_SIZE)
	if err != nil {
		log.Fatalln(err)
	}
	// ensure table is present
	var tableNames []string
	cur, err = r.DB(bmo.database).TableList().Run(session)
	if err != nil {
		log.Fatalln(err)
	}
	cur.All(&tableNames)
	set := make(map[string]bool)
	for _, v := range tableNames {
		set[v] = true
	}
	if !set[bmo.table] {
		log.Println("Creating table ", bmo.table)
		_, err = r.DB(bmo.database).TableCreate(bmo.table).RunWrite(session)
		if err != nil {
			log.Fatalln("Error creating table: ", err)
			os.Exit(1)
		}
	}

	// deliver the messages
	decoder := json.NewDecoder(input)
	ms := make([]Message, INSERT_BATCH_SIZE)
	var m *Message
	var i uint64
	var ignoreLast bool

	pool, _ := tunny.CreatePoolGeneric(POOL_SIZE).Open()
	defer pool.Close()

	table := r.Table(bmo.table)
	insertOptions := r.InsertOpts{Durability: "soft"}

	insert := func() {
		j := i
		if !ignoreLast {
			j += 1
		}
		_, err = table.Insert(ms[:j], insertOptions).RunWrite(session)
		if err != nil {
			log.Fatal(err)
			os.Exit(1)
		}
	}

	for {
		i = bmo.seq % INSERT_BATCH_SIZE
		m = &ms[i]
		err = decoder.Decode(&m)

		switch {
		case err == io.EOF:
			ignoreLast = true
			pool.SendWork(insert)
			return
		case err != nil:
			ignoreLast = true
			pool.SendWork(insert)
			log.Fatal("Can't parse json input, \"", err, "\". Object #", bmo.seq, ", after ", m)
			os.Exit(1)
		default:
			if i+1 == INSERT_BATCH_SIZE {
				ignoreLast = false
				pool.SendWork(insert)
			}
		}

		bmo.seq += 1
	}
}