func (gcmConnector *GCMConnector) broadcastMessage(msg server.MsgAndRoute) {
	topic := msg.Message.Path
	payload := gcmConnector.parseMessageToMap(msg.Message)
	guble.Info("broadcasting message with topic %v ...", string(topic))

	subscriptions := gcmConnector.kvStore.Iterate(GCM_REGISTRATIONS_SCHEMA, "")
	count := 0
	for {
		select {
		case entry, ok := <-subscriptions:
			if !ok {
				guble.Info("send message to %v receivers", count)
				return
			}
			gcmId := entry[0]
			//TODO collect 1000 gcmIds and send them in one request!
			broadcastMessage := gcm.NewMessage(payload, gcmId)
			go func() {
				//TODO error handling of response!
				_, err := gcmConnector.sender.Send(broadcastMessage, 3)
				guble.Debug("sent broadcast message to gcmId=%v", gcmId)
				if err != nil {
					guble.Err("error sending broadcast message to cgmid=%v: %v", gcmId, err.Error())
				}
			}()
			count++
		}
	}
}

func (service *Service) Stop() error {
	errors := make(map[string]error)
	for _, stopable := range service.stopListener {
		name := reflect.TypeOf(stopable).String()
		stoppedChan := make(chan bool)
		errorChan := make(chan error)
		guble.Info("stopping %v ...", name)
		go func() {
			err := stopable.Stop()
			if err != nil {
				errorChan <- err
				return
			}
			stoppedChan <- true
		}()
		select {
		case err := <-errorChan:
			guble.Err("error while stopping %v: %v", name, err.Error)
			errors[name] = err
		case <-stoppedChan:
			guble.Info("stopped %v", name)
		case <-time.After(service.StopGracePeriod):
			errors[name] = fmt.Errorf("error while stopping %v: not returned after %v seconds", name, service.StopGracePeriod)
			guble.Err(errors[name].Error())
		}
	}
	if len(errors) > 0 {
		return fmt.Errorf("Errors while stopping modules %q", errors)
	}
	return nil
}

func waitForTermination(callback func()) {
	sigc := make(chan os.Signal)
	signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM)
	guble.Info("Got singal '%v' .. exit greacefully now", <-sigc)
	callback()
	guble.Info("exit now")
	os.Exit(0)
}

func DefaultConnectionFactory(url string, origin string) (WSConnection, error) {
	guble.Info("connecting to %v", url)
	header := http.Header{"Origin": []string{origin}}
	conn, _, err := websocket.DefaultDialer.Dial(url, header)
	if err != nil {
		return nil, err
	}
	guble.Info("connected to %v", url)
	return conn, nil
}

// Registers the supplied module on this service.
// This method checks the module for the following interfaces and
// does the expected tegistrations:
//   Stopable: notify when the service stops
//   Endpoint: Register the handler function of the Endpoint in the http service at prefix
//   SetRouter: Provide the router
//   SetMessageEntry: Provide the message entry
//
// If the module does not have a HandlerFunc, the prefix parameter is ignored
func (service *Service) Register(module interface{}) {
	name := reflect.TypeOf(module).String()

	switch m := module.(type) {
	case Stopable:
		guble.Info("register %v as StopListener", name)
		service.AddStopListener(m)
	}

	switch m := module.(type) {
	case Startable:
		guble.Info("register %v as StartListener", name)
		service.AddStartListener(m)
	}

	switch m := module.(type) {
	case Endpoint:
		guble.Info("register %v as Endpoint to %v", name, m.GetPrefix())
		service.AddHandler(m.GetPrefix(), m)
	}

	// do the injections ...

	switch m := module.(type) {
	case SetKVStore:
		guble.Debug("inject KVStore to %v", name)
		m.SetKVStore(service.kvStore)
	}

	switch m := module.(type) {
	case SetMessageStore:
		guble.Debug("inject MessageStore to %v", name)
		m.SetMessageStore(service.messageStore)
	}

	switch m := module.(type) {
	case SetRouter:
		guble.Debug("inject Router to %v", name)
		m.SetRouter(service.router)
	}

	switch m := module.(type) {
	case SetMessageEntry:
		guble.Debug("inject MessageEntry to %v", name)
		m.SetMessageEntry(service.messageSink)
	}

	switch m := module.(type) {
	case SetAccessManager:
		guble.Debug("inject AccessManager to %v", name)
		m.SetAccessManager(service.accessManager)
	}
}

func (am RestAccessManager) AccessAllowed(accessType AccessType, userId string, path guble.Path) bool {

	u, _ := url.Parse(string(am))
	q := u.Query()
	if accessType == READ {
		q.Set("type", "read")
	} else {
		q.Set("type", "write")
	}

	q.Set("userId", userId)
	q.Set("path", string(path))

	resp, err := http.DefaultClient.Get(u.String())

	if err != nil {
		guble.Warn("RestAccessManager: %v", err)
		return false
	}
	defer resp.Body.Close()
	responseBody, err := ioutil.ReadAll(resp.Body)

	if err != nil || resp.StatusCode != 200 {
		guble.Info("error getting permission", err)
		guble.Debug("error getting permission", responseBody)
		return false
	}

	guble.Debug("RestAccessManager: %v, %v, %v, %v", accessType, userId, path, string(responseBody))

	return "true" == string(responseBody)

}

func (gcmConnector *GCMConnector) sendMessageToGCM(msg server.MsgAndRoute) {
	gcmId := msg.Route.ApplicationId

	payload := gcmConnector.parseMessageToMap(msg.Message)

	var messageToGcm = gcm.NewMessage(payload, gcmId)
	guble.Info("sending message to %v ...", gcmId)
	result, err := gcmConnector.sender.Send(messageToGcm, 5)
	if err != nil {
		guble.Err("error sending message to cgmid=%v: %v", gcmId, err.Error())
		return
	}

	errorJson := result.Results[0].Error
	if errorJson != "" {
		gcmConnector.handleJsonError(errorJson, gcmId, msg.Route)
	} else {
		guble.Debug("delivered message to gcm cgmid=%v: %v", gcmId, errorJson)
	}

	//we only send to one receiver, so we know that we can replace the old id with the first registration id (=canonical id)
	if result.CanonicalIDs != 0 {
		gcmConnector.replaceSubscriptionWithCanonicalID(msg.Route, result.Results[0].RegistrationID)
	}
}

func (srv *WSHandler) receiveLoop() {
	var message []byte
	for {
		err := srv.clientConn.Receive(&message)
		if err != nil {
			guble.Info("applicationId=%v closed the connection", srv.applicationId)
			srv.cleanAndClose()
			break
		}

		//guble.Debug("websocket_connector, raw message received: %v", string(message))
		cmd, err := guble.ParseCmd(message)
		if err != nil {
			srv.sendError(guble.ERROR_BAD_REQUEST, "error parsing command. %v", err.Error())
			continue
		}
		switch cmd.Name {
		case guble.CMD_SEND:
			srv.handleSendCmd(cmd)
		case guble.CMD_RECEIVE:
			srv.handleReceiveCmd(cmd)
		case guble.CMD_CANCEL:
			srv.handleCancelCmd(cmd)
		default:
			srv.sendError(guble.ERROR_BAD_REQUEST, "unknown command %v", cmd.Name)
		}
	}
}

func (gcmConnector *GCMConnector) subscribe(topic string, userid string, gcmid string) {
	guble.Info("gcm connector registration to userid=%q, gcmid=%q: %q", userid, gcmid, topic)

	route := server.NewRoute(topic, gcmConnector.channelFromRouter, gcmid, userid)

	gcmConnector.router.Subscribe(route)
	gcmConnector.saveSubscription(userid, topic, gcmid)
}

func (gcmConnector *GCMConnector) replaceSubscriptionWithCanonicalID(route *server.Route, newGcmId string) {
	oldGcmId := route.ApplicationId
	topic := string(route.Path)
	userId := route.UserId

	guble.Info("replacing old gcmId %v with canonicalId %v", oldGcmId, newGcmId)
	gcmConnector.removeSubscription(route, oldGcmId)
	gcmConnector.subscribe(topic, userId, newGcmId)
}

func (ws *WebServer) Start() error {
	guble.Info("starting up at %v", ws.addr)
	ws.server = &http.Server{Addr: ws.addr, Handler: ws.mux}
	var err error
	ws.ln, err = net.Listen("tcp", ws.addr)
	if err != nil {
		return err
	}

	go func() {
		err = ws.server.Serve(tcpKeepAliveListener{ws.ln.(*net.TCPListener)})

		if err != nil && !strings.HasSuffix(err.Error(), "use of closed network connection") {
			guble.Err("ListenAndServe %s", err.Error())
		}
		guble.Info("http server stopped")
	}()
	return nil
}

func (srv *WSHandler) cleanAndClose() {
	guble.Info("closing applicationId=%v", srv.applicationId)

	for path, rec := range srv.receiver {
		rec.Stop()
		delete(srv.receiver, path)
	}

	srv.clientConn.Close()
}

// Opens the database file.
// If the directory does not exist, it will be created.
func (kvStore *SqliteKVStore) Open() error {
	directoryPath := filepath.Dir(kvStore.filename)
	if err := ensureWriteableDirectory(directoryPath); err != nil {
		guble.Err("error db directory not writeable %q: %q", kvStore.filename, err)
		return err
	}

	guble.Info("opening sqldb %v", kvStore.filename)
	gormdb, err := gorm.Open("sqlite3", kvStore.filename)
	if err != nil {
		guble.Err("error opening sqlite3 db %q: %q", kvStore.filename, err)
		return err
	}

	if err := gormdb.DB().Ping(); err != nil {
		guble.Err("error pinging database %q: %q", kvStore.filename, err.Error())
	} else {
		guble.Debug("can ping database %q", kvStore.filename)
	}

	//gormdb.LogMode(true)
	gormdb.DB().SetMaxIdleConns(2)
	gormdb.DB().SetMaxOpenConns(5)
	gormdb.SingularTable(true)

	if err := gormdb.AutoMigrate(&kvEntry{}).Error; err != nil {
		guble.Err("error in schema migration: %q", err)
		return err
	} else {
		guble.Debug("ensured db schema")
	}

	if !kvStore.syncOnWrite {
		guble.Info("setting db: PRAGMA synchronous = OFF")
		if err := gormdb.Exec("PRAGMA synchronous = OFF").Error; err != nil {
			guble.Err("error setting PRAGMA synchronous = OFF: %v", err)
			return err
		}
	}
	kvStore.db = &gormdb
	return nil
}

func (router *PubSubRouter) unsubscribe(r *Route) {
	guble.Info("unsubscribe applicationId=%v, path=%v", r.ApplicationId, r.Path)
	routeList, present := router.routes[r.Path]
	if !present {
		return
	}
	router.routes[r.Path] = remove(routeList, r)
	if len(router.routes[r.Path]) == 0 {
		delete(router.routes, r.Path)
	}
}

func (router *PubSubRouter) deliverMessage(route Route, message *guble.Message) {
	defer guble.PanicLogger()
	select {
	case route.C <- MsgAndRoute{Message: message, Route: &route}:
		// fine, we could send the message
	default:
		guble.Info("queue was full, closing delivery for route=%v to applicationId=%v", route.Path, route.ApplicationId)
		close(route.C)
		router.unsubscribe(&route)
	}
}

func (srv *WSHandler) handleReceiveCmd(cmd *guble.Cmd) {
	rec, err := NewReceiverFromCmd(srv.applicationId, cmd, srv.sendChannel, srv.messageSouce, srv.messageStore)

	if err != nil {
		guble.Info("client error in handleReceiveCmd: %v", err.Error())
		srv.sendError(guble.ERROR_BAD_REQUEST, err.Error())
		return
	}
	srv.receiver[rec.path] = rec
	rec.Start()
}

func (router *PubSubRouter) handleMessage(message *guble.Message) {
	if guble.InfoEnabled() {
		guble.Info("routing message: %v", message.MetadataLine())
	}

	for currentRoutePath, currentRouteList := range router.routes {
		if matchesTopic(message.Path, currentRoutePath) {
			for _, route := range currentRouteList {
				router.deliverMessage(route, message)
			}
		}
	}
}

func (router *PubSubRouter) subscribe(r *Route) {
	guble.Info("subscribe applicationId=%v, path=%v", r.ApplicationId, r.Path)

	routeList, present := router.routes[r.Path]
	if !present {
		routeList = []Route{}
		router.routes[r.Path] = routeList
	}

	// try to remove, to avoid double subscriptions of the same app
	routeList = remove(routeList, r)

	router.routes[r.Path] = append(routeList, *r)
}

func (srv *WSHandler) sendLoop() {
	for {
		select {
		case raw := <-srv.sendChannel:
			if guble.DebugEnabled() {
				if len(raw) < 80 {
					guble.Debug("send to client (userId=%v, applicationId=%v, totalSize=%v): %v", srv.userId, srv.applicationId, len(raw), string(raw))
				} else {
					guble.Debug("send to client (userId=%v, applicationId=%v, totalSize=%v): %v...", srv.userId, srv.applicationId, len(raw), string(raw[0:79]))
				}
			}
			if err := srv.clientConn.Send(raw); err != nil {
				guble.Info("applicationId=%v closed the connection", srv.applicationId)
				srv.cleanAndClose()
				break
			}
		}
	}
}

func (gcmConnector *GCMConnector) loadSubscriptions() {
	subscriptions := gcmConnector.kvStore.Iterate(GCM_REGISTRATIONS_SCHEMA, "")
	count := 0
	for {
		select {
		case entry, ok := <-subscriptions:
			if !ok {
				guble.Info("renewed %v gcm subscriptions", count)
				return
			}
			gcmId := entry[0]
			splitedValue := strings.SplitN(entry[1], ":", 2)
			userid := splitedValue[0]
			topic := splitedValue[1]

			guble.Debug("renew gcm subscription: user=%v, topic=%v, gcmid=%v", userid, topic, gcmId)
			route := server.NewRoute(topic, gcmConnector.channelFromRouter, gcmId, userid)
			gcmConnector.router.Subscribe(route)
			count++
		}
	}
}