// ZMQ_MULTICAST_HOPS: Maximum network hops for multicast packets
//
// See: http://api.zeromq.org/4-1:zmq-setsockopt#toc21
509215
func (soc *Socket) SetMulticastHops(value int) error {
	return soc.setInt(C.ZMQ_MULTICAST_HOPS, value)
}
1035286
func (soc *Socket) SetReqRelaxed(value int) error {
	return soc.setInt(C.ZMQ_REQ_RELAXED, value)
}
1199248
func (s *Session) SetMaxIdleConns(n int) {
	s.mu.Lock()
	defer s.mu.Unlock()

	s.opts.MaxIdle = n
	s.cluster.SetMaxIdleConns(n)
}
895193
func ZmqRequestSocket(ip string, request_ports []int) *zmq.Socket {
	context, _ := zmq.NewContext()
	socket, _ := context.NewSocket(zmq.REQ)
	for _, _port := range request_ports {
		socket.Connect(fmt.Sprintf("tcp://%s:%d", ip, _port))
	}
	return socket
}
1224596
func (wsh *WebsocketHub) MaybeSendTxns(tx *types.MempoolTx) {
	if wsh.addTxToBuffer(tx) {
		// This is called from the event loop, so these sends channel may not be
		// blocking.
		go func() {
			wsh.bufferTickerChan <- tickerSigReset
			wsh.sendBufferChan <- bufferSend
		}()
	}
}
1298796
func (o *MultiCastChannelMap) MultiCastRanges(info *bambou.FetchingInfo) (MultiCastRangesList, *bambou.Error) {

	var list MultiCastRangesList
	err := bambou.CurrentSession().FetchChildren(o, MultiCastRangeIdentity, &list, info)
	return list, err
}
788304
func hasMultipleServers(config clientcmdapi.Config) bool {
	server := ""
	for _, cluster := range config.Clusters {
		if len(server) == 0 {
			server = cluster.Server
		}
		if server != cluster.Server {
			return true
		}
	}
	return false
}
775890
func NewAMQPConnection(host string) (*amqp.Connection, *amqp.Channel) {
	connection, err := amqp.Dial(host)
	if err != nil {
		panic(err)
	}

	channel, err := connection.Channel()
	if err != nil {
		panic(err)
	}
	return connection, channel
}
970453
func ScheduleMulticastQueueBatch(size int) error {
	return storage.Transaction(func(tx sqlx.Ext) error {
		// this locks the selected queue-items so that this query can be
		// executed by other instances in parallel.
		multicastQueueItems, err := storage.GetSchedulableMulticastQueueItems(tx, size)
		if err != nil {
			return errors.Wrap(err, "get multicast queue-items error")
		}

		for _, qi := range multicastQueueItems {
			err := multicast.HandleScheduleQueueItem(tx, qi)
			if err != nil {
				log.WithFields(log.Fields{
					"multicast_group_id": qi.MulticastGroupID,
					"id":                 qi.ID,
				}).WithError(err).Error("schedule multicast queue-item error")
			}
		}

		return nil
	})
}
716478
func (c *Connection) setSocketOptions() {
	c.ws.SetReadLimit(c.MaxMessageSize)
	c.keepAlive()
	c.ws.SetPongHandler(func(string) error {
		c.log("Received Pong from Client", LogLevelDebug)
		c.keepAlive()
		return nil
	})
}
341838
func StringCast(s string) Multiaddr {
	m, err := NewMultiaddr(s)
	if err != nil {
		panic(fmt.Errorf("multiaddr failed to parse: %s", err))
	}
	return m
}
865243
func ZmqProxySocket(proxy ZmqProxyConfig) {
	chanSource := make(chan []byte, 5)
	chanDestination := make(chan []byte, 5)

	go proxyDestination(proxy.DestinationIP, proxy.DestinationPorts, chanSource, chanDestination)
	go proxySource(proxy.SourceIP, proxy.SourcePorts, chanSource, chanDestination)
}
1211761
func (conR *ConsensusReactor) queryMaj23Routine(peer p2p.Peer, ps *PeerState) {
	logger := conR.Logger.With("peer", peer)

OUTER_LOOP:
	for {
		// Manage disconnects from self or peer.
		if !peer.IsRunning() || !conR.IsRunning() {
			logger.Info("Stopping queryMaj23Routine for peer")
			return
		}

		// Maybe send Height/Round/Prevotes
		{
			rs := conR.conS.GetRoundState()
			prs := ps.GetRoundState()
			if rs.Height == prs.Height {
				if maj23, ok := rs.Votes.Prevotes(prs.Round).TwoThirdsMajority(); ok {
					peer.TrySend(StateChannel, cdc.MustMarshalBinaryBare(&VoteSetMaj23Message{
						Height:  prs.Height,
						Round:   prs.Round,
						Type:    types.PrevoteType,
						BlockID: maj23,
					}))
					time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)
				}
			}
		}

		// Maybe send Height/Round/Precommits
		{
			rs := conR.conS.GetRoundState()
			prs := ps.GetRoundState()
			if rs.Height == prs.Height {
				if maj23, ok := rs.Votes.Precommits(prs.Round).TwoThirdsMajority(); ok {
					peer.TrySend(StateChannel, cdc.MustMarshalBinaryBare(&VoteSetMaj23Message{
						Height:  prs.Height,
						Round:   prs.Round,
						Type:    types.PrecommitType,
						BlockID: maj23,
					}))
					time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)
				}
			}
		}

		// Maybe send Height/Round/ProposalPOL
		{
			rs := conR.conS.GetRoundState()
			prs := ps.GetRoundState()
			if rs.Height == prs.Height && prs.ProposalPOLRound >= 0 {
				if maj23, ok := rs.Votes.Prevotes(prs.ProposalPOLRound).TwoThirdsMajority(); ok {
					peer.TrySend(StateChannel, cdc.MustMarshalBinaryBare(&VoteSetMaj23Message{
						Height:  prs.Height,
						Round:   prs.ProposalPOLRound,
						Type:    types.PrevoteType,
						BlockID: maj23,
					}))
					time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)
				}
			}
		}

		// Little point sending LastCommitRound/LastCommit,
		// These are fleeting and non-blocking.

		// Maybe send Height/CatchupCommitRound/CatchupCommit.
		{
			prs := ps.GetRoundState()
			if prs.CatchupCommitRound != -1 && 0 < prs.Height && prs.Height <= conR.conS.blockStore.Height() {
				commit := conR.conS.LoadCommit(prs.Height)
				peer.TrySend(StateChannel, cdc.MustMarshalBinaryBare(&VoteSetMaj23Message{
					Height:  prs.Height,
					Round:   commit.Round(),
					Type:    types.PrecommitType,
					BlockID: commit.BlockID,
				}))
				time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)
			}
		}

		time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)

		continue OUTER_LOOP
	}
}
602199
func (c *MConnection) sendSomePacketMsgs() bool {
	// Block until .sendMonitor says we can write.
	// Once we're ready we send more than we asked for,
	// but amortized it should even out.
	c.sendMonitor.Limit(c._maxPacketMsgSize, atomic.LoadInt64(&c.config.SendRate), true)

	// Now send some PacketMsgs.
	for i := 0; i < numBatchPacketMsgs; i++ {
		if c.sendPacketMsg() {
			return true
		}
	}
	return false
}
1112909
func (soc *Socket) SetRouterRaw(value int) error {
	return soc.setInt(C.ZMQ_ROUTER_RAW, value)
}
1054097
func NewMultiplexTransport(
	nodeInfo NodeInfo,
	nodeKey NodeKey,
	mConfig conn.MConnConfig,
) *MultiplexTransport {
	return &MultiplexTransport{
		acceptc:          make(chan accept),
		closec:           make(chan struct{}),
		dialTimeout:      defaultDialTimeout,
		filterTimeout:    defaultFilterTimeout,
		handshakeTimeout: defaultHandshakeTimeout,
		mConfig:          mConfig,
		nodeInfo:         nodeInfo,
		nodeKey:          nodeKey,
		conns:            NewConnSet(),
		resolver:         net.DefaultResolver,
	}
}
110597
func (n *NetworkServerAPI) FlushMulticastQueueForMulticastGroup(ctx context.Context, req *ns.FlushMulticastQueueForMulticastGroupRequest) (*empty.Empty, error) {
	var mgID uuid.UUID
	copy(mgID[:], req.MulticastGroupId)

	if err := storage.FlushMulticastQueueForMulticastGroup(storage.DB(), mgID); err != nil {
		return nil, errToRPCError(err)
	}

	return &empty.Empty{}, nil
}