gotask.go

package main

import (
	"fmt"
	"gopkg.in/mgo.v2"
	"gopkg.in/mgo.v2/bson"
	"os"
	"github.com/bitly/go-nsq"
	"log"
)

type User struct {
	Id    bson.ObjectId `json:"id,omitempty" bson:"_id,omitempty"`
	Email string `bson:"email"`
}

type ProducerTransaction struct {
    Error error         // the error (or nil) of the publish command
    Args  []interface{} // the slice of variadic arguments passed to PublishAsync or MultiPublishAsync
    // contains filtered or unexported fields
}

func main() {
	uri := os.Getenv("MONGOHQ_URL")
	if uri == "" {
		fmt.Println("no connection string provided")
		os.Exit(1)
	}

	sess, err := mgo.Dial(uri)
	if err != nil {
		fmt.Printf("Can't connect to mongo, go error %v\n", err)
		os.Exit(1)
	}
	defer sess.Close()

	sess.SetSafe(&mgo.Safe{})

	var users []User
	err = sess.DB("db_name").C("users").Find(bson.M{}).All(&users)

	if err != nil {
		os.Exit(1)
	}

	// connect to nsq
	config := nsq.NewConfig()
	w, _ := nsq.NewProducer("HOST", config)


	// push each user into nsq
	var producer_t nsq.ProducerTransaction
	for _, user := range users {
		err = w.PublishAsync("write_test", []byte(user.Id), &producer_t)
		if err != nil {
			log.Panic("Could not connect")
		}
	}

	// w.MultiPublishAsync

	w.Stop()

}