// initSubscribers performs one-time initialization of the Postgres listener and
// goroutine for event delivery, termination and subscription management.
func (n *Notifier) initSubscribers(dbUrl string) error {
	n.l = pq.NewListener(dbUrl, 20*time.Millisecond, time.Hour, nil)
	go func() {
		for {
			select {
			case <-n.terminated:
				return
			case <-n.exit:
				n.l.UnlistenAll()
				n.l.Close()
				n.db.Close()
				for _, localL := range n.subscribers {
					close(localL)
				}
				close(n.terminated)
			case cmd := <-n.ch:
				cmd.fun()
				close(cmd.ack)
			case pgn := <-n.l.Notify:
				if pgn != nil {
					localN, err := n.makeNotification(pgn)
					if err != nil {
						log.Printf("Error parsing inbound notification %v: %v", pgn, err)
					} else {
						for _, sub := range n.subscribers {
							sub <- localN
						}
					}
				}
			}
		}
	}()

	return nil
}

func waitForNotification(dbcluter DBCluster, parition string) {
	conninfo := fmt.Sprintf("postgres://%s:%[email protected]%s/%s?sslmode=require", "read", dbcluter.Password, dbcluter.Parition[parition].Write.IP, "regentmarkets")
	listener := pq.NewListener(conninfo, 5*time.Second, 10*time.Second, nil)
	db, _ := sql.Open("postgres", conninfo)
	err := db.Ping()
	if err != nil {
		log.Fatal(err)
	}
	err = listener.Listen("transactions_watcher")
	checkErr(err)
	fmt.Println("Listing to", parition)

	var redisdb redis.Conn
	var notification *pq.Notification
	for {
		select {
		case notification = <-listener.Notify:
			if notification != nil {
				redisconn(&redisdb)
				publish(redisdb, regexp.MustCompile(",").Split(notification.Extra, -1))
			}

		case <-time.After(60 * time.Second):
			fmt.Println("no notifications for 60 seconds...")
		}
	}
}

func (cmd *activityCmd) run(ctx scope.Context, args []string) error {
	// Get cluster in order to load config.
	_, err := getCluster(ctx)
	if err != nil {
		return fmt.Errorf("cluster error: %s", err)
	}

	listener := pq.NewListener(backend.Config.DB.DSN, 200*time.Millisecond, 5*time.Second, nil)
	if err := listener.Listen("broadcast"); err != nil {
		return fmt.Errorf("pq listen error: %s", err)
	}

	defer func() {
		ctx.Cancel()
		ctx.WaitGroup().Wait()
	}()

	// Start metrics server.
	ctx.WaitGroup().Add(1)
	go activity.Serve(ctx, cmd.addr)

	// Start scanner.
	ctx.WaitGroup().Add(1)
	activity.ScanLoop(ctx, listener)

	return nil
}

func main() {
	conninfo := "dbname=gotraining user=root password=root port=7705 sslmode=disable"

	_, err := sql.Open("postgres", conninfo)
	if err != nil {
		panic(err)
	}

	reportProblem := func(ev pq.ListenerEventType, err error) {
		if err != nil {
			fmt.Println(err.Error())
		}
	}

	listener := pq.NewListener(conninfo, 10*time.Second, time.Minute, reportProblem)
	err = listener.Listen("events")
	if err != nil {
		panic(err)
	}

	fmt.Println("Start monitoring PostgreSQL...")
	for {
		waitForNotification(listener)
	}
}

func Init() {
	var err error

	db, err = sql.Open("postgres", *dataSource)
	if err != nil {
		debug.Fatal(err)
	}

	create := func(name, content string) {
		if err != nil {
			debug.Fatal(err)
		}
		err = createTable(name, content)
	}

	// primary tables
	create("user_", createUserSQL)
	create("project", createProjectSQL)
	create("task", createTaskSQL)
	create("milestone", createMilestoneSQL)
	create("member", createMemberSQL)
	create("worker", createWorkerSQL)
	create("friend", createFriendSQL)
	create("chat", createChatSQL)

	// secondary tables
	create("featured_project", createFeaturedProjectSQL)
	create("user_tag", createUserTagSQL)
	create("task_tag", createTaskTagSQL)
	create("tag", createTagSQL)

	// setup listener
	if *listenerEnabled {
		listener = pq.NewListener(*dataSource, 1*time.Second, time.Minute, func(ev pq.ListenerEventType, err error) {
			if err != nil {
				log.Fatal(err)
			}
		})

		if err := listener.Listen("chat"); err != nil {
			log.Fatal(err)
		}

		go func() {
			for {
				select {
				case notification := <-listener.Notify:
					if Notify != nil {
						Notify(notification.Channel, notification.Extra)
					}
				}
			}
		}()
	} else {
		debug.Warn("PostgreSQL listener is disabled")
	}
}

func listenMessages() {
	revel.TRACE.Println("Setting up db listening")
	var Spec, _ = revel.Config.String("db.spec")
	var listener = pq.NewListener(Spec, 10*time.Second, time.Minute, listenerEventsHandler)
	var err = listener.Listen("new_message")
	if err != nil {
		panic(err)
	}
	waitForNotification(listener)
}

func (repo *postgresRepository) ensureListener() {
	repo.m.Lock()
	defer repo.m.Unlock()

	if repo.listener == nil {
		repo.listener = pq.NewListener(repo.connectionString, time.Second, time.Minute, nil)

		kit.SafeGo(repo.listenDaemon)
	}
}

func run(config Config) {
	listener := pq.NewListener(config.PostgresURL, 10*time.Second, time.Minute, errorReporter)
	err := listener.Listen("urlwork")
	if err != nil {
		log.Fatal(err)
	}

	rabbitchannel := make(chan string, 100)

	go func() {
		cfg := new(tls.Config)
		cfg.InsecureSkipVerify = true
		conn, err := amqp.DialTLS(config.RabbitMQURL, cfg)
		if err != nil {
			log.Fatal(err)
		}
		defer conn.Close()

		ch, err := conn.Channel()
		if err != nil {
			log.Fatal(err)
		}
		defer ch.Close()

		for {
			payload := <-rabbitchannel
			log.Println(payload)
			err := ch.Publish("urlwork", "todo", false, false, amqp.Publishing{
				ContentType: "text/plain",
				Body:        []byte(payload),
			})
			if err != nil {
				log.Fatal(err)
			}
		}
	}()

	for {
		select {
		case notification := <-listener.Notify:
			rabbitchannel <- notification.Extra
		case <-time.After(90 * time.Second):
			go func() {
				err := listener.Ping()
				if err != nil {
					log.Fatal(err)
				}
			}()
		}
	}
}

// newPgListener creates and returns the pglistener from the pq package.
func newPgListener(pgconninfo string) (*pq.Listener, error) {

	// create a callback function to monitor connection state changes
	pgEventCallback := func(ev pq.ListenerEventType, err error) {
		if err != nil {
			fmt.Println("pgbroadcast: ", err.Error())
		}
	}

	// create the listener
	l := pq.NewListener(pgconninfo, 10*time.Second, time.Minute, pgEventCallback)

	return l, nil
}

func run(config Config) {
	purl := fmt.Sprintf("user=%s password=%s host=%s port=%s dbname=%s sslmode=%s", config.DbUser, config.DbPassword, config.DbHost, config.DbPort, config.DbName, config.SslMode)
	listener := pq.NewListener(purl, 10*time.Second, time.Minute, errorReporter)
	err := listener.Listen("usertrigger")
	if err != nil {
		log.Fatal(err)
	}

	rabbitchannel := make(chan string, 100)

	//Code for STOMP
	go func() {
		rabbitHost := fmt.Sprintf("%s:%s", config.RabbitHost, config.RabbitPort)
		conn, err := stomp.Dial("tcp", rabbitHost,
			stomp.ConnOpt.Login(config.RabbitUser, config.RabbitPassword),
			stomp.ConnOpt.AcceptVersion(stomp.V11),
			stomp.ConnOpt.AcceptVersion(stomp.V12),
			stomp.ConnOpt.Host(config.RabbitVHost),
			stomp.ConnOpt.Header("nonce", "B256B26D320A"))

		if err != nil {
			log.Fatal(err)
		}
		defer conn.Disconnect()

		for {
			payload := <-rabbitchannel
			log.Println(payload)
			err = conn.Send(config.RabbitQueue, "text/plain", []byte(payload))
			if err != nil {
				log.Fatal(err)
			}
		}
	}()

	for {
		select {
		case notification := <-listener.Notify:
			rabbitchannel <- notification.Extra
		case <-time.After(90 * time.Second):
			go func() {
				err := listener.Ping()
				if err != nil {
					log.Fatal(err)
				}
			}()
		}
	}
}

func (jql *jobQueueListener) background(wg *sync.WaitGroup) {
	ctx := jql.Backend.ctx.Fork()
	logger := jql.Backend.logger

	defer ctx.WaitGroup().Done()

	listener := pq.NewListener(jql.Backend.dsn, 200*time.Millisecond, 5*time.Second, nil)
	if err := listener.Listen("job_item"); err != nil {
		// TODO: manage this more nicely
		panic("job listen: " + err.Error())
	}
	logger.Printf("job listener started")

	// Signal to constructor that we're ready to handle operations.
	wg.Done()

	keepalive := time.NewTicker(30 * time.Second)
	defer keepalive.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-keepalive.C:
			// Ping to make sure the database connection is still live.
			if err := listener.Ping(); err != nil {
				logger.Printf("job listener ping: %s\n", err)
				jql.Backend.ctx.Terminate(fmt.Errorf("job listener ping: %s", err))
				return
			}
		case notice := <-listener.Notify:
			if notice == nil {
				logger.Printf("job listener: received nil notification")
				// A nil notice indicates a loss of connection.
				// For now it's easier to just shut down and force job
				// processor to restart.
				jql.Backend.ctx.Terminate(ErrPsqlConnectionLost)
				return
			}

			jql.m.Lock()
			if c, ok := jql.cs[notice.Extra]; ok {
				c.Signal()
			}
			jql.m.Unlock()
		}
	}
}

func (cmd *ATCCommand) constructDB(logger lager.Logger) (*db.SQLDB, db.PipelineDBFactory, error) {
	dbConn, err := migrations.LockDBAndMigrate(logger.Session("db.migrations"), "postgres", cmd.PostgresDataSource)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to migrate database: %s", err)
	}

	listener := pq.NewListener(cmd.PostgresDataSource, time.Second, time.Minute, nil)
	bus := db.NewNotificationsBus(listener, dbConn)

	explainDBConn := db.Explain(logger, dbConn, clock.NewClock(), 500*time.Millisecond)
	sqlDB := db.NewSQL(logger.Session("db"), explainDBConn, bus)

	pipelineDBFactory := db.NewPipelineDBFactory(logger.Session("db"), explainDBConn, bus, sqlDB)

	return sqlDB, pipelineDBFactory, err
}

func waitForNotification(dbcluter DBCluster, parition string) {
	conninfo := fmt.Sprintf("postgres://%s:%[email protected]%s/%s?sslmode=require", "read", dbcluter.Password, dbcluter.Parition[parition].Write.IP, "regentmarkets")
	listener := pq.NewListener(conninfo, 5*time.Second, 10*time.Second, nil)
	_ = listener.Listen("transactions_watcher")

	redisdb, _ := redis.DialURL(os.Getenv("REDIS_URL"))
	var notification *pq.Notification
	for {
		select {
		case notification = <-listener.Notify:
			if notification != nil {
				publish(redisdb, regexp.MustCompile(",").Split(notification.Extra, -1))
			}
		}
	}
}

// Listen executes `LISTEN channel`. Uses f to handle received notifications on chanel.
// On error logs error messages (if a logs exists)
func (db *Database) Listen(channel string, f func(payload ...string)) error {
	// Create a new listener only if Listen is called for the first time
	if db.listener == nil {
		db.listenerCallbacks = make(map[string]func(...string))

		reportProblem := func(ev pq.ListenerEventType, err error) {
			if err != nil && db.logger != nil {
				db.printLog(err.Error())
			}
		}
		db.listener = pq.NewListener(db.connectionString, 10*time.Second, time.Minute, reportProblem)

		if db.listener == nil {
			return errors.New("Unable to create a new listener")
		}

		// detach event handler
		go func() {
			for {
				select {
				case notification := <-db.listener.Notify:
					go db.listenerCallbacks[notification.Channel](notification.Extra)
				case <-time.After(90 * time.Second):
					go func() {
						if db.listener.Ping() != nil {
							db.printLog(fmt.Sprintf("Error checking server connection for channel %s\n", channel))
							return
						}
					}()
				}
			}
		}()
	}

	if _, alreadyIn := db.listenerCallbacks[channel]; alreadyIn {
		return errors.New("Already subscribed to channel " + channel)
	}

	db.listenerCallbacks[channel] = f

	if err := db.listener.Listen(channel); err != nil {
		return err
	}

	return nil
}

// pgstr - строка для коннекта к постгресу
func NewEventService(pgstr string) (event *EventService) {

	event = &EventService{Channels: map[string]Channel{}}

	reportProblem := func(ev pq.ListenerEventType, err error) {
		if err != nil {
			event.handleErrorAll(err)
		}
	}

	event.Listener = pq.NewListener(pgstr, 10*time.Second, time.Minute, reportProblem)

	go func() {
		var msg Message
		for {
			select {
			case n := <-event.Listener.Notify:
				if ch, ok := event.Channels[n.Channel]; ok {
					ch.MessageCount++
					err := json.Unmarshal([]byte(n.Extra), &msg)
					// не распарсили сообщение
					if err != nil {
						ch.handleError(err)
						continue
					}
					// распарсили и нашли клиента
					if client, ok := ch.Clients[msg.Target]; ok {
						go client.EventHandler(msg.Body)
					}
				}
				continue
			case <-time.After(60 * time.Second):
				go func() {
					err := event.Listener.Ping()
					if err != nil {
						event.handleErrorAll(err)
					}
				}()
				continue
			}
		}
	}()

	return
}

func waitForNotification(clientdb ClientDB, company string) {
	conninfo := fmt.Sprintf("postgres://%s:%[email protected]%s/%s?sslmode=require", "postgres", clientdb.Password, clientdb.Company[company].Write.IP, "test")
	listener := pq.NewListener(conninfo, 10*time.Second, time.Minute, nil)
	err := listener.Listen("getwork")
	checkErr(err)

	redisdb, err := redis.DialURL(os.Getenv("REDIS_URL"))
	checkErr(err)
	var notification *pq.Notification
	for {
		select {
		case notification = <-listener.Notify:
			if notification != nil {
				publish(redisdb, regexp.MustCompile(",").Split(notification.Extra, -1))
			}

		case <-time.After(60 * time.Second):
			fmt.Println("no notifications for 60 seconds...")
		}
	}
}

func NewDispatcher(conf DBConf) (d *Dispatcher, err error) {
	c, ok := conf[env()]
	if !ok {
		err = errors.New("Configuration for \"" + env() + "\" environment not found.")
		return
	}

	d = &Dispatcher{
		done:      make(chan bool),
		waitGroup: &sync.WaitGroup{},
		listener:  pq.NewListener(c.Conn, 10*time.Second, time.Minute, nil),
		clients:   NewClientsPool(),
		tasks:     make(map[int64]autogram.BackgroundTask),
	}

	d.db, err = sql.Open(c.Driver, c.Conn)
	if err != nil {
		return
	}
	return
}

func main() {
	config_dsn := os.Getenv("CP_DSN")
	if config_dsn == "" {
		config_dsn = "postgres://cachepusher:[email protected]:5432/cachepusher?sslmode=disable"
	}

	config_redis := os.Getenv("CP_REDIS")
	if config_redis == "" {
		config_redis = "10.10.42.23:6379"
	}

	config_cacheprefix := os.Getenv("CP_PREFIX")
	if config_cacheprefix == "" {
		config_cacheprefix = "customer"
	}

	db, err := sqlx.Connect("postgres", config_dsn)
	if err != nil {
		log.Print("[Postgresql] ", err)
		return
	}

	r := redis.NewClient(&redis.Options{
		Addr: config_redis,
	})
	_, err = r.Ping().Result()
	if err != nil {
		log.Print("[Redis] ", err)
		return
	}

	log.Print("Clearing cache")
	keys, err := r.Keys(fmt.Sprintf("%s:*", config_cacheprefix)).Result()
	if err != nil {
		log.Print(err)
	}
	r.Pipelined(func(r *redis.Pipeline) error {
		for _, key := range keys {
			err = r.Del(key).Err()
			if err != nil {
				log.Print(err)
			}
		}
		return nil
	})

	listener := pq.NewListener(config_dsn, 10*time.Second, time.Minute, func(ev pq.ListenerEventType, err error) {
		if err != nil {
			log.Print(err)
		}
	})

	err = listener.Listen("customer_updated")
	if err != nil {
		log.Print(err)
		return
	}
	err = listener.Listen("customer_deleted")
	if err != nil {
		log.Print(err)
		return
	}

	/** Wait until we have set up the listener to get notifications before we trigger the warmup **/
	log.Print("Triggering cache warmup")
	_, err = db.Exec("select customer_warmup()")
	if err != nil {
		log.Print(err)
	}

	log.Printf("Listening for updates")
	for {
		select {
		case n := <-listener.Notify:
			cachekey := fmt.Sprintf("%s:%s", config_cacheprefix, n.Extra)
			switch n.Channel {
			case "customer_deleted":
				log.Printf("DEL %s", cachekey)
				err = r.Del(cachekey).Err()
				if err != nil {
					log.Print(err)
				}
			case "customer_updated":
				var customerJson string
				err = db.Get(&customerJson, "select customer_get($1)", n.Extra)
				if err != nil {
					log.Print(err)
				}
				log.Printf("SET %s %s", cachekey, customerJson)
				err = r.Set(cachekey, customerJson, 0).Err()
				if err != nil {
					log.Print(err)
				}
			}
		// Make sure our connection stays up
		case <-time.After(90 * time.Second):
			log.Print("LISTEN PING")
			go func() {
				err = listener.Ping()
				if err != nil {
//.........这里部分代码省略.........

func main() {
	flag.Parse()

	if !*dev && (*httpUsername == "" || (*httpHashedPassword == "" && *httpPassword == "")) {
		fatal(errors.New("must specify -httpUsername and -httpPassword or -httpHashedPassword or turn on dev mode"))
	}

	if _, err := os.Stat(*templatesDir); err != nil {
		fatal(errors.New("directory specified via -templates does not exist"))
	}

	if _, err := os.Stat(*publicDir); err != nil {
		fatal(errors.New("directory specified via -public does not exist"))
	}

	logger := lager.NewLogger("atc")

	logLevel := lager.INFO
	if *dev {
		logLevel = lager.DEBUG
	}

	sink := lager.NewReconfigurableSink(lager.NewWriterSink(os.Stdout, lager.DEBUG), logLevel)
	logger.RegisterSink(sink)

	var err error

	var dbConn Db.Conn

	for {
		dbConn, err = migration.Open(*sqlDriver, *sqlDataSource, migrations.Migrations)
		if err != nil {
			if strings.Contains(err.Error(), " dial ") {
				logger.Error("failed-to-open-db", err)
				time.Sleep(5 * time.Second)
				continue
			}

			fatal(err)
		}

		break
	}

	dbConn = Db.Explain(logger, dbConn, 500*time.Millisecond)

	listener := pq.NewListener(*sqlDataSource, time.Second, time.Minute, nil)
	bus := Db.NewNotificationsBus(listener)

	db := Db.NewSQL(logger.Session("db"), dbConn, bus)
	pipelineDBFactory := Db.NewPipelineDBFactory(logger.Session("db"), dbConn, bus, db)

	var configDB Db.ConfigDB
	configDB = Db.PlanConvertingConfigDB{db}

	var resourceTypesNG []atc.WorkerResourceType
	err = json.Unmarshal([]byte(*resourceTypes), &resourceTypesNG)
	if err != nil {
		logger.Fatal("invalid-resource-types", err)
	}

	var workerClient worker.Client
	if *gardenAddr != "" {
		workerClient = worker.NewGardenWorker(
			gclient.New(gconn.NewWithLogger(
				*gardenNetwork,
				*gardenAddr,
				logger.Session("garden-connection"),
			)),
			clock.NewClock(),
			-1,
			resourceTypesNG,
			"linux",
			[]string{},
		)
	} else {
		workerClient = worker.NewPool(worker.NewDBWorkerProvider(db, logger))
	}

	resourceTracker := resource.NewTracker(workerClient)
	gardenFactory := exec.NewGardenFactory(workerClient, resourceTracker, func() string {
		guid, err := uuid.NewV4()
		if err != nil {
			panic("not enough entropy to generate guid: " + err.Error())
		}

		return guid.String()
	})
	execEngine := engine.NewExecEngine(gardenFactory, engine.NewBuildDelegateFactory(db), db)

	engine := engine.NewDBEngine(engine.Engines{execEngine}, db, db)

	var webValidator auth.Validator

	if *httpUsername != "" && *httpHashedPassword != "" {
		webValidator = auth.BasicAuthHashedValidator{
			Username:       *httpUsername,
			HashedPassword: *httpHashedPassword,
		}
	} else if *httpUsername != "" && *httpPassword != "" {
//.........这里部分代码省略.........

func main() {
	InitErrorLog(os.Stderr)

	if len(os.Args) != 2 {
		printUsage()
		os.Exit(1)
	} else if os.Args[1] == "--help" {
		printUsage()
		os.Exit(1)
	}

	err := readConfigFile(os.Args[1])
	if err != nil {
		elog.Fatalf("error while reading configuration file: %s", err)
	}
	if len(Config.Databases) == 0 {
		elog.Fatalf("at least one database must be configured")
	}

	listenAddr := net.JoinHostPort(Config.ListenAddress, strconv.Itoa(Config.ListenPort))
	tcpAddr, err := net.ResolveTCPAddr("tcp", listenAddr)
	if err != nil {
		elog.Fatalf("could not resolve listen address: %s", err)
	}
	l, err := net.ListenTCP("tcp", tcpAddr)
	if err != nil {
		elog.Fatalf("could not open listen socket: %s", err)
	}

	var m sync.Mutex
	var connStatusNotifier chan struct{}

	listenerStateChange := func(ev pq.ListenerEventType, err error) {
		switch ev {
		case pq.ListenerEventConnectionAttemptFailed:
			elog.Warningf("Listener: could not connect to the database: %s", err.Error())

		case pq.ListenerEventDisconnected:
			elog.Warningf("Listener: lost connection to the database: %s", err.Error())
			m.Lock()
			close(connStatusNotifier)
			connStatusNotifier = nil
			m.Unlock()

		case pq.ListenerEventReconnected,
			pq.ListenerEventConnected:
			elog.Logf("Listener: connected to the database")
			m.Lock()
			connStatusNotifier = make(chan struct{})
			m.Unlock()
		}
	}

	// make sure pq.Listener doesn't pick up any env variables
	os.Clearenv()

	clientConnectionString := fmt.Sprintf("fallback_application_name=allas %s", Config.ClientConnInfo)
	listener := pq.NewListener(clientConnectionString,
		250*time.Millisecond, 5*time.Minute,
		listenerStateChange)

	nd := notifydispatcher.NewNotifyDispatcher(listener)
	nd.SetBroadcastOnConnectionLoss(false)
	nd.SetSlowReaderEliminationStrategy(notifydispatcher.NeglectSlowReaders)

	for {
		c, err := l.Accept()
		if err != nil {
			panic(err)
		}

		var myConnStatusNotifier chan struct{}

		m.Lock()
		if connStatusNotifier == nil {
			m.Unlock()
			go RejectFrontendConnection(c)
			continue
		} else {
			myConnStatusNotifier = connStatusNotifier
		}
		m.Unlock()

		newConn := NewFrontendConnection(c, nd, myConnStatusNotifier)
		go newConn.mainLoop(Config.StartupParameters, Config.Databases)
	}
}