| 70 | } |
| 71 | |
| 72 | func New(id string, cfg *Config) (*Postgresql, error) { |
| 73 | if len(id) == 0 { |
| 74 | return nil, errors.New("id must NOT be a empty string") |
| 75 | } |
| 76 | |
| 77 | connStr := fmt.Sprintf("postgres://%s:%s@%s/%s?sslmode=disable", cfg.Username, cfg.Password, cfg.Addrs[0], cfg.DBName) |
| 78 | db, err := sql.Open("postgres", connStr) |
| 79 | if err != nil { |
| 80 | return nil, err |
| 81 | } |
| 82 | |
| 83 | listener := pq.NewListener(connStr, listenerMinReconnectInterval, listenerMaxReconnectInterval, nil) |
| 84 | err = listener.Listen(cfg.NotifyChannel) |
| 85 | if err != nil { |
| 86 | return nil, err |
| 87 | } |
| 88 | |
| 89 | electPath := defaultElectPath |
| 90 | if cfg.ElectPath != "" { |
| 91 | electPath = defaultElectPath |
| 92 | } |
| 93 | |
| 94 | p := &Postgresql{ |
| 95 | myID: id, |
| 96 | electPath: electPath, |
| 97 | db: db, |
| 98 | listener: listener, |
| 99 | quitCh: make(chan struct{}), |
| 100 | lockReleaseCh: make(chan bool), |
| 101 | leaderChangeCh: make(chan bool), |
| 102 | } |
| 103 | err = p.initLeaderId() |
| 104 | if err != nil { |
| 105 | return nil, err |
| 106 | } |
| 107 | p.isReady.Store(false) |
| 108 | p.wg.Add(2) |
| 109 | go p.electLoop() |
| 110 | go p.observeLeaderEvent() |
| 111 | return p, nil |
| 112 | } |
| 113 | |
| 114 | func (p *Postgresql) ID() string { |
| 115 | return p.myID |