diff --git a/config/config.go b/config/config.go index ae98753..a762807 100644 --- a/config/config.go +++ b/config/config.go @@ -47,6 +47,7 @@ type Storage struct { // Config represents a configuration for a server.Server. type Config struct { Addr string `json:"addr"` + PubAddr string `json:"pub_addr"` Storage Storage `json:"storage"` Private bool `json:"private"` diff --git a/config/config_test.go b/config/config_test.go index 7447271..d969ebd 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -16,6 +16,7 @@ var exampleJson = `{ "network": "tcp", "addr": ":34000", + "pub_addr": "tcp://*:34001", "storage": { "driver": "redis", "addr": "127.0.0.1:6379", diff --git a/server/publish.go b/server/publish.go new file mode 100644 index 0000000..f1af5cf --- /dev/null +++ b/server/publish.go @@ -0,0 +1,28 @@ +// Copyright 2013 The Chihaya Authors. All rights reserved. +// Use of this source code is governed by the BSD 2-Clause license, +// which can be found in the LICENSE file. +package server + +import ( + zmq "github.com/alecthomas/gozmq" +) + +func (s *Server) publish() { + context, err := zmq.NewContext() + if err != nil { + panic(err) + } + defer context.Close() + + socket, err := context.NewSocket(zmq.PUB) + if err != nil { + panic(err) + } + defer socket.Close() + + socket.Bind(s.conf.PubAddr) + + for msg := range s.pubChan { + socket.Send([]byte(msg), 0) + } +} diff --git a/server/server.go b/server/server.go index 7b7d893..e775bda 100644 --- a/server/server.go +++ b/server/server.go @@ -34,6 +34,8 @@ type Server struct { waitgroup sync.WaitGroup + pubChan chan string + http.Server } @@ -46,6 +48,7 @@ func New(conf *config.Config) (*Server, error) { s := &Server{ conf: conf, dbConnPool: pool, + pubChan: make(chan string), Server: http.Server{ Addr: conf.Addr, ReadTimeout: conf.ReadTimeout.Duration, @@ -66,6 +69,7 @@ func (s *Server) ListenAndServe() error { s.startTime = time.Now() go s.updateRPM() + go s.publish() s.Serve(s.listener) s.waitgroup.Wait() @@ -79,6 +83,7 @@ func (s *Server) Stop() error { if err != nil { return err } + close(s.pubChan) return s.listener.Close() } diff --git a/vagrant/puppet/manifests/default.pp b/vagrant/puppet/manifests/default.pp index 8fe001d..b876016 100644 --- a/vagrant/puppet/manifests/default.pp +++ b/vagrant/puppet/manifests/default.pp @@ -3,33 +3,54 @@ exec { 'echo this works': } group { 'puppet': ensure => 'present' } -exec { 'chown -R vagrant:vagrant /home/vagrant/': } +exec { 'chown -R vagrant:vagrant /home/vagrant/': +} exec { 'apt-get update': command => '/usr/bin/apt-get update', } +exec { 'add-apt-repository ppa:chris-lea/zeromq && apt-get update': + require => Package['python-software-properties'], + alias => 'zmq_repo', + creates => '/etc/apt/sources.list.d/chris-lea-zeromq-precise.list', +} + exec { 'add-apt-repository ppa:duh/golang && apt-get update': alias => 'go_repo', creates => '/etc/apt/sources.list.d/gophers-go-precise.list', require => Package['python-software-properties'], } -package { 'python-software-properties': - ensure => present, +package { 'pkg-config': require => Exec['apt-get update'], + ensure => present, +} + +package { 'libzmq-dev': + require => [ + Exec['zmq_repo'], + Package['pkg-config'], + ], + ensure => present, +} + +package { 'python-software-properties': + require => Exec['apt-get update'], + ensure => present, } package { 'git': - ensure => present, require => Exec['apt-get update'], + ensure => present, } package { 'golang': - ensure => present, require => Exec['go_repo'], + ensure => present, } exec { 'echo "export GOPATH=/home/vagrant/chihaya" > /etc/profile.d/gopath.sh': + alias => 'go_path', creates => '/etc/profile.d/gopath.sh', }