// CreateTopic request topic creation
190875
func (b *Broker) CreateTopic(topics []proto.TopicInfo, timeout time.Duration, validateOnly bool) (*proto.CreateTopicsResp, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	var resp *proto.CreateTopicsResp
	err := b.callOnClusterController(func(c *connection) error {
		var err error
		req := proto.CreateTopicsReq{
			Timeout:              timeout,
			CreateTopicsRequests: topics,
			ValidateOnly:         validateOnly,
		}
		resp, err = c.CreateTopic(&req)
		return err
	})
	return resp, err
}
452600
func ensureTopic(pubsubClient *pubsub.Client, topicName string) *pubsub.Topic {
	var topic *pubsub.Topic
	ctx, _ := context.WithTimeout(context.Background(), ContextDuration)
	topic = pubsubClient.Topic(topicName)
	topicExists, err := topic.Exists(ctx)

	if err != nil {
		log.Fatalf("Could not check if topic exists: %v", err)
	}
	if !topicExists {
		new_topic, err := pubsubClient.CreateTopic(ctx, topicName)
		if err != nil {
			log.Fatalf("Could not create PubSub topic: %v", err)
		}
		topic = new_topic
	}

	return topic
}
637680
func (s *NotificationConfigurationDeprecated) SetTopicConfiguration(v *TopicConfigurationDeprecated) *NotificationConfigurationDeprecated {
	s.TopicConfiguration = v
	return s
}
895201
func (l *Client) GetTopicSummary(topic string, limit int) (TopicSummary, error) {
	res := ApiResp{}
	data := TopicSummary{}
	params := url.Values{}

	if limit > 0 {
		params.Add("limit", strconv.Itoa(limit))
	}

	err := l.Get(parseLyticsURL(topicEndpoint, map[string]string{"topicId": topic}), params, nil, &res, &data)
	if err != nil {
		return data, err
	}

	return data, nil
}
528250
func (l *Logger) SetTopic(topic string) {
	if len(topic) == 0 {
		panic("Empty tag")
	}

	l.topic.Store(topic)
}
256180
func (fs *Filters) addTopicMatcher(watcher *Filter) {
	if len(watcher.Topics) == 0 {
		fs.allTopicsMatcher[watcher] = struct{}{}
	} else {
		for _, t := range watcher.Topics {
			topic := BytesToTopic(t)
			if fs.topicMatcher[topic] == nil {
				fs.topicMatcher[topic] = make(map[*Filter]struct{})
			}
			fs.topicMatcher[topic][watcher] = struct{}{}
		}
	}
}
1289354
func (this *ZookeeperCoordinator) GetAllTopics() (topics []string, err error) {
	backoffMultiplier := 1
	for i := 0; i <= this.config.MaxRequestRetries; i++ {
		topics, err = this.tryGetAllTopics()
		if err == nil {
			return
		}
		Tracef(this, "GetAllTopics failed after %d-th retry", i)
		time.Sleep(this.config.RequestBackoff * time.Duration(backoffMultiplier))
		backoffMultiplier++
	}
	return
}
1122060
func Create(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Cache-Control", "must-revalidate")
	w.Header().Set("Content-Type", "text/plain")
	w.Header().Set("Access-Control-Allow-Origin", "*")
	w.Header().Set("Access-Control-Allow-Methods", "POST, GET, PUT, OPTIONS")
	var col string
	if !Require(w, r, "col", &col) {
		return
	}
	if err := HttpDB.Create(col); err != nil {
		http.Error(w, fmt.Sprint(err), http.StatusBadRequest)
	} else {
		w.WriteHeader(http.StatusCreated)
	}
}
103033
func (e *Events) SubscribeTopic(since, until time.Time, ef *Filter) ([]eventtypes.Message, chan interface{}) {
	eventSubscribers.Inc()
	e.mu.Lock()

	var topic func(m interface{}) bool
	if ef != nil && ef.filter.Len() > 0 {
		topic = func(m interface{}) bool { return ef.Include(m.(eventtypes.Message)) }
	}

	buffered := e.loadBufferedEvents(since, until, topic)

	var ch chan interface{}
	if topic != nil {
		ch = e.pub.SubscribeTopic(topic)
	} else {
		// Subscribe to all events if there are no filters
		ch = e.pub.Subscribe()
	}

	e.mu.Unlock()
	return buffered, ch
}
386136
func createRequest() *http.Request {
	// Create HTTP request
	req := &http.Request{
		Method:     "GET",
		URL:        &url.URL{},
		Host:       "",
		ProtoMajor: 1,
		ProtoMinor: 1,
		Proto:      "HTTP/1.1",
		Header:     make(http.Header),
		Body:       utils.NopCloser(),
	}
	// Return shallow copy of Request with the new context
	return req.WithContext(emptyContext())
}
311846
func createRequestURI(base string, params url.Values) string {
	if len(params) != 0 {
		base += "?" + params.Encode()
	}
	return base
}
141454
func (c *client) createRequest(rawurl, method string, in interface{}) (*http.Request, error) {
	uri, err := url.Parse(rawurl)
	if err != nil {
		return nil, err
	}

	// if we are posting or putting data, we need to
	// write it to the body of the request.
	var buf io.ReadWriter
	if in != nil {
		buf = new(bytes.Buffer)
		err := json.NewEncoder(buf).Encode(in)
		if err != nil {
			return nil, err
		}
	}

	// creates a new http request to bitbucket.
	req, err := http.NewRequest(method, uri.String(), buf)
	if err != nil {
		return nil, err
	}
	if in != nil {
		req.Header.Set("Content-Type", "application/json")
	}
	return req, nil
}
326004
func (ctl *taskController) PrepareTopic(ctx context.Context, publisher string) (topic string, token string, err error) {
	return ctl.eng.prepareTopic(ctx, &topicParams{
		jobID:     ctl.JobID(),
		invID:     ctl.InvocationID(),
		manager:   ctl.manager,
		publisher: publisher,
	})
}
911731
func (i *IssuesService) Create(uri *Hyperlink, uriParams M, requestParams interface{}) (issue *Issue, result *Result) {
	url, err := ExpandWithDefault(uri, &RepoIssuesURL, uriParams)
	if err != nil {
		return nil, &Result{Err: err}
	}

	result = i.client.post(url, requestParams, &issue)
	return
}
368451
func GetTopic(e interface{}) string {
	switch e.(type) {
	case *events.TaskCreate:
		return runtime.TaskCreateEventTopic
	case *events.TaskStart:
		return runtime.TaskStartEventTopic
	case *events.TaskOOM:
		return runtime.TaskOOMEventTopic
	case *events.TaskExit:
		return runtime.TaskExitEventTopic
	case *events.TaskDelete:
		return runtime.TaskDeleteEventTopic
	case *events.TaskExecAdded:
		return runtime.TaskExecAddedEventTopic
	case *events.TaskExecStarted:
		return runtime.TaskExecStartedEventTopic
	case *events.TaskPaused:
		return runtime.TaskPausedEventTopic
	case *events.TaskResumed:
		return runtime.TaskResumedEventTopic
	case *events.TaskCheckpointed:
		return runtime.TaskCheckpointedEventTopic
	default:
		logrus.Warnf("no topic for type %#v", e)
	}
	return runtime.TaskUnknownTopic
}
581822
func (s *TestRoleInput) SetTopics(v []*string) *TestRoleInput {
	s.Topics = v
	return s
}
409155
func (r *httpVPackRequest) createHTTPRequest(endpoint url.URL) (*http.Request, error) {
	r.written = false
	u := endpoint
	u.Path = ""
	url := u.String()
	if !strings.HasSuffix(url, "/") {
		url = url + "/"
	}
	p := r.path
	if strings.HasPrefix(p, "/") {
		p = p[1:]
	}
	url = url + p
	if r.q != nil {
		q := r.q.Encode()
		if len(q) > 0 {
			url = url + "?" + q
		}
	}
	var body io.Reader
	if r.body != nil {
		body = bytes.NewReader(r.body)
	}
	req, err := http.NewRequest(r.method, url, body)
	if err != nil {
		return nil, driver.WithStack(err)
	}

	if r.hdr != nil {
		for k, v := range r.hdr {
			req.Header.Set(k, v)
		}
	}

	req.Header.Set("Accept", "application/x-velocypack")
	//req.Header.Set("Accept", "application/json")
	if r.body != nil {
		req.Header.Set("Content-Length", strconv.Itoa(len(r.body)))
		req.Header.Set("Content-Type", "application/x-velocypack")
	}
	return req, nil
}