func (topics *Topics) DeleteTopic(cfg *Config, name string) bool {
	client := cfg.RiakConnection()
	bucket, err := client.NewBucketType("maps", "config")
	topicsConfig, err := bucket.FetchMap("topicsConfig")
	if err != nil {
		logrus.Error(err)
	}
	topicsConfig.FetchSet("topics").Remove([]byte(name))
	err = topicsConfig.Store()
	if err != nil {
		logrus.Error(err)
	}
	topics.Lock()
	topics.TopicMap[name].Delete(cfg)
	delete(topics.TopicMap, name)
	topics.Unlock()
	if err != nil {
		logrus.Error(err)
		return false
	}
	return true
}
DeleteTopic will delete the topic from the collection of all topics, which
removes any queues it's subscription list
package app
import (
	"fmt"
	"sync"
	"time"
	"github.com/Sirupsen/logrus"
	"github.com/tpjg/goriakpbc"
)
type Topic struct {
	Name     string
	Config   *riak.RDtMap
	riakPool *riak.Client
	queues   *Queues
	sync.RWMutex
}
type Topics struct {
	Config *riak.RDtMap
	TopicMap map[string]*Topic
	riakPool *riak.Client
	queues   *Queues
	syncScheduler *time.Ticker
	syncKiller    chan struct{}
	sync.RWMutex
}
func InitTopics(cfg *Config, queues *Queues) *Topics {
	client := cfg.RiakConnection()
	bucket, err := client.NewBucketType("maps", "config")
	if err != nil {
		logrus.Error(err)
	}
	config, err := bucket.FetchMap("topicsConfig")
	if err != nil {
		logrus.Error(err)
	}
	if config.FetchSet("topics") == nil {
		topicSet := config.AddSet("topics")
		topicSet.Add([]byte("default_topic"))
		err = config.Store()
	}
	if err != nil {
		logrus.Error(err)
	}
	topics := Topics{
		Config:   config,
		riakPool: cfg.RiakPool,
		queues:   queues,
		TopicMap: make(map[string]*Topic),
	}
	go topics.scheduleSync(cfg)
	return &topics
}
func (topics *Topics) InitTopic(name string) {
	client := topics.riakPool
	bucket, _ := client.NewBucketType("maps", ConfigurationBucket)
	config, _ := bucket.FetchMap(topicConfigRecordName(name))
	topic := new(Topic)
	topic.Config = config
	topic.Name = name
	topic.riakPool = topics.riakPool
	topic.queues = topics.queues
	topics.TopicMap[name] = topic
	topic.Config.Store()
	topics.Config.FetchSet("topics").Add([]byte(name))
	topics.Config.Store()
}
func (topic *Topic) Broadcast(cfg *Config, message string) map[string]string {
	queueWrites := make(map[string]string)
	topicQueues := topic.getConfig().FetchSet("queues")
	if topicQueues != nil {
		for _, queue := range topicQueues.GetValue() {
			var present bool
			_, present = topic.queues.QueueMap[string(queue)]
			if present == true {
				uuid := topic.queues.QueueMap[string(queue)].Put(cfg, message)
				queueWrites[string(queue)] = uuid
			} else {
			}
		}
	}
	return queueWrites
}
func (topic *Topic) AddQueue(cfg *Config, name string) {
	client := cfg.RiakConnection()
	bucket, err := client.NewBucketType("maps", ConfigurationBucket)
	recordName := topicConfigRecordName(topic.Name)
	topic.Config, err = bucket.FetchMap(recordName)
	queueSet := topic.Config.AddSet("queues")
	queueSet.Add([]byte(name))
	topic.Config.Store()
	topic.Config, err = bucket.FetchMap(recordName)
	if err != nil {
		logrus.Error(err)
	}
}
func (topic *Topic) DeleteQueue(cfg *Config, name string) {
	client := cfg.RiakConnection()
	recordName := topicConfigRecordName(topic.Name)
	bucket, _ := client.NewBucketType("maps", ConfigurationBucket)
	topic.Config, _ = bucket.FetchMap(recordName)
	topic.Config.FetchSet("queues").Remove([]byte(name))
	topic.Config.Store()
	topic.Config, _ = bucket.FetchMap(recordName)
}
func (topic *Topic) ListQueues() []string {
	list := make([]string, 0, 10)
	queueList := topic.Config.FetchSet("queues")
	if queueList != nil {
		for _, queueName := range queueList.GetValue() {
			list = append(list, string(queueName))
		}
	}
	return list
}