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 (c *Client) startWithReconnect() {
	for {
		if c.Connected {
			err := c.readLoop()
			if err == nil {
				return
			}
		}

		if c.shouldStop() {

			return
		}

		var err error
		c.ws, err = c.WSConnectionFactory(c.url, c.origin)
		if err != nil {
			c.Connected = false
			guble.Err("error on connect, retry in 50ms: %v", err)
			time.Sleep(time.Millisecond * 50)
		} else {
			c.Connected = true
			guble.Err("connected again")
		}
	}
}

func StartupService(args Args) *server.Service {
	router := server.NewPubSubRouter()
	router.SetAccessManager(server.NewAllowAllAccessManager(true))
	router.Go()
	service := server.NewService(
		args.Listen,
		CreateKVStoreBackend(args),
		CreateMessageStoreBackend(args),
		server.NewMessageEntry(router),
		router,
		server.NewAllowAllAccessManager(true))

	for _, module := range CreateModules(args) {
		service.Register(module)
	}

	if err := service.Start(); err != nil {
		guble.Err(err.Error())
		if err := service.Stop(); err != nil {
			guble.Err(err.Error())
		}
		os.Exit(1)
	}

	return service
}

func (gcmConnector *GCMConnector) handleJsonError(jsonError string, gcmId string, route *server.Route) {
	if jsonError == "NotRegistered" {
		guble.Debug("remove not registered cgm registration cgmid=%v", gcmId)
		gcmConnector.removeSubscription(route, gcmId)
	} else if jsonError == "InvalidRegistration" {
		guble.Err("the cgmid=%v is not registered. %v", gcmId, jsonError)
	} else {
		guble.Err("unexpected error while sending to cgm cgmid=%v: %v", gcmId, jsonError)
	}
}

func (c *Client) handleIncommoingMessage(msg []byte) {
	parsed, err := guble.ParseMessage(msg)
	if err != nil {
		guble.Err("parsing message failed %v", err)
		c.errors <- clientErrorMessage(err.Error())
		return
	}

	switch message := parsed.(type) {
	case *guble.Message:
		c.messages <- message
	case *guble.NotificationMessage:
		if message.IsError {
			select {
			case c.errors <- message:
			default:
			}
		} else {
			select {
			case c.statusMessages <- message:
			default:
			}
		}
	}
}

func writeLoop(client *client.Client) {
	shouldStop := false
	for !shouldStop {
		func() {
			defer guble.PanicLogger()
			reader := bufio.NewReader(os.Stdin)
			text, _ := reader.ReadString('\n')
			if strings.TrimSpace(text) == "" {
				return
			}

			if strings.TrimSpace(text) == "?" || strings.TrimSpace(text) == "help" {
				printHelp()
				return
			}

			if strings.HasPrefix(text, ">") {
				fmt.Print("header: ")
				header, _ := reader.ReadString('\n')
				text += header
				fmt.Print("body: ")
				body, _ := reader.ReadString('\n')
				text += strings.TrimSpace(body)
			}

			if args.Verbose {
				log.Printf("Sending: %v\n", text)
			}
			if err := client.WriteRawMessage([]byte(text)); err != nil {
				shouldStop = true
				guble.Err(err.Error())
			}
		}()
	}
}

func Main() {
	defer func() {
		if p := recover(); p != nil {
			guble.Err("%v", p)
			os.Exit(1)
		}
	}()

	args := loadArgs()
	if args.LogInfo {
		guble.LogLevel = guble.LEVEL_INFO
	}
	if args.LogDebug {
		guble.LogLevel = guble.LEVEL_DEBUG
	}

	if err := ValidateStoragePath(args); err != nil {
		os.Exit(1)
	}

	service := StartupService(args)

	waitForTermination(func() {
		service.Stop()
	})
}

func (rec *Receiver) fetchOnlyLoop() {
	err := rec.fetch()
	if err != nil {
		guble.Err("error while fetching: %v, %+v", err.Error(), rec)
		rec.sendError(guble.ERROR_INTERNAL_SERVER, err.Error())
	}
}

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 (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++
		}
	}
}

// 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 (fms *FileMessageStore) Stop() error {
	fms.mutex.Lock()
	defer fms.mutex.Unlock()

	var returnError error
	for key, partition := range fms.partitions {
		if err := partition.Close(); err != nil {
			returnError = err
			guble.Err("error on closing message store partition %q: %v", key, err)
		}
		delete(fms.partitions, key)
	}
	return returnError
}

// Take the message and forward it to the router.
func (entry *MessageEntry) HandleMessage(msg *guble.Message) error {
	txCallback := func(msgId uint64) []byte {
		msg.Id = msgId
		msg.PublishingTime = time.Now().Format(time.RFC3339)
		return msg.Bytes()
	}

	if err := entry.messageStore.StoreTx(msg.Path.Partition(), txCallback); err != nil {
		guble.Err("error storing message in partition %v: %v", msg.Path.Partition(), err)
		return err
	}

	return entry.router.HandleMessage(msg)
}

func (service *Service) Start() error {
	el := guble.NewErrorList("Errors occured while startup the service: ")

	for _, startable := range service.startListener {
		name := reflect.TypeOf(startable).String()

		guble.Debug("starting module %v", name)
		if err := startable.Start(); err != nil {
			guble.Err("error on startup module %v", name)
			el.Add(err)
		}
	}
	return el.ErrorOrNil()
}

func (rec *Receiver) subscriptionLoop() {
	for !rec.shouldStop {
		if rec.doFetch {

			if err := rec.fetch(); err != nil {
				guble.Err("error while fetching: %v, %+v", err.Error(), rec)
				rec.sendError(guble.ERROR_INTERNAL_SERVER, err.Error())
				return
			}

			if err := rec.messageStore.DoInTx(rec.path.Partition(), rec.subscribeIfNoUnreadMessagesAvailable); err != nil {
				if err == unread_messages_available {
					//fmt.Printf(" unread_messages_available lastSendId=%v\n", rec.lastSendId)
					rec.startId = int64(rec.lastSendId + 1)
					continue // fetch again
				} else {
					guble.Err("error while subscribeIfNoUnreadMessagesAvailable: %v, %+v", err.Error(), rec)
					rec.sendError(guble.ERROR_INTERNAL_SERVER, err.Error())
					return
				}
			}
		}

		if !rec.doFetch {
			rec.subscribe()
		}
		rec.receiveFromSubscription()

		if !rec.shouldStop {
			//fmt.Printf(" router closed .. on msg: %v\n", rec.lastSendId)
			// the router kicked us out, because we are to slow for realtime listening,
			// so we setup parameters for fetching and cloging the gap. Than we can subscribe again.
			rec.startId = int64(rec.lastSendId + 1)
			rec.doFetch = true
		}
	}
}

func (c *Client) readLoop() error {
	for {
		if _, msg, err := c.ws.ReadMessage(); err != nil {
			c.Connected = false
			if c.shouldStop() {
				return nil
			} else {
				guble.Err("read error: %v", err.Error())
				c.errors <- clientErrorMessage(err.Error())
				return err
			}
		} else {
			guble.Debug("raw> %s", msg)
			c.handleIncommoingMessage(msg)
		}
	}
}

func TestStopingOfModulesTimeout(t *testing.T) {
	defer initCtrl(t)()

	// given:
	service, _, _, _, _ := aMockedService()
	service.StopGracePeriod = time.Millisecond * 5

	// whith a registered stopable, which blocks to long on stop
	stopable := NewMockStopable(ctrl)
	service.Register(stopable)
	stopable.EXPECT().Stop().Do(func() {
		time.Sleep(time.Millisecond * 10)
	})

	// then the Stop returns with an error
	err := service.Stop()
	assert.Error(t, err)
	guble.Err(err.Error())
}

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 (kvStore *SqliteKVStore) IterateKeys(schema string, keyPrefix string) chan string {
	responseChan := make(chan string, 100)

	go func() {

		rows, err := kvStore.db.Raw("select key from kv_entry where schema = ? and key LIKE ?", schema, keyPrefix+"%").
			Rows()

		if err != nil {
			guble.Err("error fetching keys from db %v", err)
		} else {
			defer rows.Close()
			for rows.Next() {
				var value string
				rows.Scan(&value)
				responseChan <- value
			}
		}
		close(responseChan)
	}()
	return responseChan
}