// Following are actions from the parent grpc.ClientConn, forward to sub-balancers.
// SubConn state change: find the corresponding balancer and then forward.
174052
func (bg *balancerGroup) handleSubConnStateChange(sc balancer.SubConn, state connectivity.State) {
	grpclog.Infof("balancer group: handle subconn state change: %p, %v", sc, state)
	bg.mu.Lock()
	var b balancer.Balancer
	if id, ok := bg.scToID[sc]; ok {
		if state == connectivity.Shutdown {
			// Only delete sc from the map when state changed to Shutdown.
			delete(bg.scToID, sc)
		}
		b = bg.idToBalancer[id]
	}
	bg.mu.Unlock()
	if b == nil {
		grpclog.Infof("balancer group: balancer not found for sc state change")
		return
	}
	if ub, ok := b.(balancer.V2Balancer); ok {
		ub.UpdateSubConnState(sc, balancer.SubConnState{ConnectivityState: state})
	} else {
		b.HandleSubConnStateChange(sc, state)
	}
}
868977
func main() {
	//  Start child threads
	go publisher_thread()
	go subscriber_thread()
	go listener_thread()

	time.Sleep(100 * time.Millisecond)

	subscriber, _ := zmq.NewSocket(zmq.XSUB)
	subscriber.Connect("tcp://localhost:6000")
	publisher, _ := zmq.NewSocket(zmq.XPUB)
	publisher.Bind("tcp://*:6001")
	listener, _ := zmq.NewSocket(zmq.PAIR)
	listener.Connect("inproc://pipe")
	zmq.Proxy(subscriber, publisher, listener)

	fmt.Println("interrupted")
}
357467
func NewClientConn(addr string, opts ...grpc.DialOption) *grpc.ClientConn {
	return NewClientConnWithContext(context.Background(), addr, opts...)
}
1102533
func (s *Switch) forward(packet *htlcPacket) error {
	switch htlc := packet.htlc.(type) {
	case *lnwire.UpdateAddHTLC:
		circuit := newPaymentCircuit(&htlc.PaymentHash, packet)
		actions, err := s.circuits.CommitCircuits(circuit)
		if err != nil {
			log.Errorf("unable to commit circuit in switch: %v", err)
			return err
		}

		// Drop duplicate packet if it has already been seen.
		switch {
		case len(actions.Drops) == 1:
			return ErrDuplicateAdd

		case len(actions.Fails) == 1:
			if packet.incomingChanID == sourceHop {
				return err
			}

			var failure lnwire.FailureMessage
			update, err := s.cfg.FetchLastChannelUpdate(
				packet.incomingChanID,
			)
			if err != nil {
				failure = &lnwire.FailTemporaryNodeFailure{}
			} else {
				failure = lnwire.NewTemporaryChannelFailure(update)
			}
			addErr := ErrIncompleteForward

			return s.failAddPacket(packet, failure, addErr)
		}

		packet.circuit = circuit
	}

	return s.route(packet)
}
949480
func (c *NodeClient) dynamicListenAndForward(ctx context.Context, ln net.Listener) {
	defer ln.Close()
	defer c.Close()

	for {
		// Accept connection from the client. Here the client is typically
		// something like a web browser or other SOCKS5 aware application.
		conn, err := ln.Accept()
		if err != nil {
			log.Errorf("Dynamic port forwarding (SOCKS5) failed: %v.", err)
			break
		}

		// Perform the SOCKS5 handshake with the client to find out the remote
		// address to proxy.
		remoteAddr, err := socks.Handshake(conn)
		if err != nil {
			log.Errorf("SOCKS5 handshake failed: %v.", err)
			break
		}
		log.Debugf("SOCKS5 proxy forwarding requests to %v.", remoteAddr)

		// Proxy the connection to the remote address.
		go func() {
			err := c.proxyConnection(ctx, conn, remoteAddr)
			if err != nil {
				log.Warnf("Failed to proxy connection: %v.", err)
			}
		}()
	}
}
543979
func (mt *membershipTracker) checkIfPeersChanged(prevPeers discovery.Members, currPeers discovery.Members,
	prevSetPeers map[string]struct{}, currSetPeers map[string]struct{}) {
	var currView [][]string

	wereInPrev := endpoints(prevPeers.Filter(func(member discovery.NetworkMember) bool {
		_, exists := currSetPeers[string(member.PKIid)]
		return !exists
	}))
	newInCurr := endpoints(currPeers.Filter(func(member discovery.NetworkMember) bool {
		_, exists := prevSetPeers[string(member.PKIid)]
		return !exists
	}))
	currView = endpoints(currPeers)

	if !reflect.DeepEqual(wereInPrev, newInCurr) {
		if len(wereInPrev) == 0 {
			mt.report("Membership view has changed. peers went online: ", newInCurr, ", current view: ", currView)
		} else if len(newInCurr) == 0 {
			mt.report("Membership view has changed. peers went offline: ", wereInPrev, ", current view: ", currView)
		} else {
			mt.report("Membership view has changed. peers went offline: ", wereInPrev, ", peers went online: ", newInCurr, ", current view: ", currView)
		}
	}
}
547591
func (c *Client) ConsumeRemoteRelationChange(change params.RemoteRelationChangeEvent) error {
	args := params.RemoteRelationsChanges{
		Changes: []params.RemoteRelationChangeEvent{change},
	}
	var results params.ErrorResults
	err := c.facade.FacadeCall("ConsumeRemoteRelationChanges", args, &results)
	if err != nil {
		return errors.Trace(err)
	}
	return results.OneError()
}
469921
func (n *NotifyRouter) HandleFSOnlineStatusChanged(online bool) {
	if n == nil {
		return
	}
	// For all connections we currently have open...
	n.cm.ApplyAll(func(id ConnectionID, xp rpc.Transporter) bool {
		// If the connection wants the `kbfs` notification type
		if n.getNotificationChannels(id).Kbfs {
			// In the background do...
			go func() {
				// A send of a `FSOnlineStatusChanged` RPC with the
				// notification
				(keybase1.NotifyFSClient{
					Cli: rpc.NewClient(xp, NewContextifiedErrorUnwrapper(n.G()), nil),
				}).FSOnlineStatusChanged(context.Background(), online)
			}()
		}
		return true
	})
	n.runListeners(func(listener NotifyListener) {
		listener.FSOnlineStatusChanged(online)
	})
}
999208
func (pf *PortForwarder) forward() error {
	var err error

	listenSuccess := false
	for i := range pf.ports {
		port := &pf.ports[i]
		err = pf.listenOnPort(port)
		switch {
		case err == nil:
			listenSuccess = true
		default:
			if pf.errOut != nil {
				fmt.Fprintf(pf.errOut, "Unable to listen on port %d: %v\n", port.Local, err)
			}
		}
	}

	if !listenSuccess {
		return fmt.Errorf("Unable to listen on any of the requested ports: %v", pf.ports)
	}

	if pf.Ready != nil {
		close(pf.Ready)
	}

	// wait for interrupt or conn closure
	select {
	case <-pf.stopChan:
	case <-pf.streamConn.CloseChan():
		runtime.HandleError(errors.New("lost connection to pod"))
	}

	return nil
}
360051
func (client *Client) StopLoadBalancerListener(loadBalancerId string, port int) (err error) {
	args := &CommonLoadBalancerListenerArgs{
		LoadBalancerId: loadBalancerId,
		ListenerPort:   port,
	}
	response := &CommonLoadBalancerListenerResponse{}
	err = client.Invoke("StopLoadBalancerListener", args, response)
	return err
}
1298273
func (c *ConnectionStore) connect(endpoint string, expectedServerCert []byte) (*grpc.ClientConn, error) {
	c.lock.Lock()
	defer c.lock.Unlock()
	// Check again to see if some other goroutine has already connected while
	// we were waiting on the lock
	conn, alreadyConnected := c.Connections.Lookup(expectedServerCert)
	if alreadyConnected {
		return conn, nil
	}

	v := c.verifyHandshake(endpoint, expectedServerCert)
	conn, err := c.dialer.Dial(endpoint, v)
	if err != nil {
		return nil, err
	}

	c.Connections.Put(expectedServerCert, conn)
	return conn, nil
}
651091
func (c *ConnectionStore) Connection(endpoint string, expectedServerCert []byte) (*grpc.ClientConn, error) {
	c.lock.RLock()
	conn, alreadyConnected := c.Connections.Lookup(expectedServerCert)
	c.lock.RUnlock()

	if alreadyConnected {
		return conn, nil
	}

	// Else, we need to connect to the remote endpoint
	return c.connect(endpoint, expectedServerCert)
}
184502
func subscriber(srv *clonesrv_t) (err error) {
	//  Get state snapshot if necessary
	if !srv.kvmap_init {
		srv.kvmap_init = true
		snapshot, _ := zmq.NewSocket(zmq.DEALER)
		snapshot.Connect(fmt.Sprint("tcp://localhost:", srv.peer))
		fmt.Printf("I: asking for snapshot from: tcp://localhost:%v\n", srv.peer)
		snapshot.SendMessage("ICANHAZ?", "") // blank subtree to get all
		for {
			kvmsg, e := kvmsg.RecvKvmsg(snapshot)
			if e != nil {
				err = e
				break
			}
			if key, _ := kvmsg.GetKey(); key == "KTHXBAI" {
				srv.sequence, _ = kvmsg.GetSequence()
				break //  Done
			}
			kvmsg.Store(srv.kvmap)
		}
		fmt.Println("I: received snapshot =", srv.sequence)
	}
	//  Find and remove update off pending list
	kvmsg, e := kvmsg.RecvKvmsg(srv.subscriber)
	if e != nil {
		err = e
		return
	}

	if key, _ := kvmsg.GetKey(); key != "HUGZ" {
		if !srv.was_pending(kvmsg) {
			//  If active update came before client update, flip it
			//  around, store active update (with sequence) on pending
			//  list and use to clear client update when it comes later
			srv.pending = append(srv.pending, kvmsg)
		}
		//  If update is more recent than our kvmap, apply it
		if seq, _ := kvmsg.GetSequence(); seq > srv.sequence {
			srv.sequence = seq
			kvmsg.Store(srv.kvmap)
			fmt.Println("I: received update =", srv.sequence)
		}
	}
	return
}
422061
func (t *Transport) ReleasePeer(pid peer.Identifier, ps peer.Subscriber) error {
	t.lock.Lock()
	defer t.lock.Unlock()
	address := pid.Identifier()
	p, ok := t.addressToPeer[address]
	if !ok {
		return peer.ErrTransportHasNoReferenceToPeer{
			TransportName:  "grpc.Transport",
			PeerIdentifier: address,
		}
	}
	if err := p.Unsubscribe(ps); err != nil {
		return err
	}
	if p.NumSubscribers() == 0 {
		delete(t.addressToPeer, address)
		p.stop()
		return p.wait()
	}
	return nil
}
1162966
func GRPCServerTrace(options ...TracerOption) kitgrpc.ServerOption {
	cfg := TracerOptions{}

	for _, option := range options {
		option(&cfg)
	}

	serverBefore := kitgrpc.ServerBefore(
		func(ctx context.Context, md metadata.MD) context.Context {
			var name string

			if cfg.Name != "" {
				name = cfg.Name
			} else {
				name, _ = ctx.Value(kitgrpc.ContextKeyRequestMethod).(string)
				if name == "" {
					// we can't find the gRPC method. probably the
					// unaryInterceptor was not wired up.
					name = "unknown grpc method"
				}
			}

			var (
				parentContext trace.SpanContext
				traceContext  = md[propagationKey]
				ok            bool
			)

			if len(traceContext) > 0 {
				traceContextBinary := []byte(traceContext[0])
				parentContext, ok = propagation.FromBinary(traceContextBinary)
				if ok && !cfg.Public {
					ctx, _ = trace.StartSpanWithRemoteParent(
						ctx,
						name,
						parentContext,
						trace.WithSpanKind(trace.SpanKindServer),
						trace.WithSampler(cfg.Sampler),
					)
					return ctx
				}
			}
			ctx, span := trace.StartSpan(
				ctx,
				name,
				trace.WithSpanKind(trace.SpanKindServer),
				trace.WithSampler(cfg.Sampler),
			)
			if ok {
				span.AddLink(
					trace.Link{
						TraceID: parentContext.TraceID,
						SpanID:  parentContext.SpanID,
						Type:    trace.LinkTypeChild,
					},
				)
			}
			return ctx
		},
	)

	serverFinalizer := kitgrpc.ServerFinalizer(
		func(ctx context.Context, err error) {
			if span := trace.FromContext(ctx); span != nil {
				if s, ok := status.FromError(err); ok {
					span.SetStatus(trace.Status{Code: int32(s.Code()), Message: s.Message()})
				} else {
					span.SetStatus(trace.Status{Code: int32(codes.Internal), Message: err.Error()})
				}
				span.End()
			}
		},
	)

	return func(s *kitgrpc.Server) {
		serverBefore(s)
		serverFinalizer(s)
	}
}
656683
func (c *ChainInfo) Forward(action Action, ip net.IP, port int, proto, destAddr string, destPort int, bridgeName string) error {
	daddr := ip.String()
	if ip.IsUnspecified() {
		// iptables interprets "0.0.0.0" as "0.0.0.0/32", whereas we
		// want "0.0.0.0/0". "0/0" is correctly interpreted as "any
		// value" by both iptables and ip6tables.
		daddr = "0/0"
	}

	args := []string{
		"-p", proto,
		"-d", daddr,
		"--dport", strconv.Itoa(port),
		"-j", "DNAT",
		"--to-destination", net.JoinHostPort(destAddr, strconv.Itoa(destPort))}
	if !c.HairpinMode {
		args = append(args, "!", "-i", bridgeName)
	}
	if err := ProgramRule(Nat, c.Name, action, args); err != nil {
		return err
	}

	args = []string{
		"!", "-i", bridgeName,
		"-o", bridgeName,
		"-p", proto,
		"-d", destAddr,
		"--dport", strconv.Itoa(destPort),
		"-j", "ACCEPT",
	}
	if err := ProgramRule(Filter, c.Name, action, args); err != nil {
		return err
	}

	args = []string{
		"-p", proto,
		"-s", destAddr,
		"-d", destAddr,
		"--dport", strconv.Itoa(destPort),
		"-j", "MASQUERADE",
	}

	if err := ProgramRule(Nat, "POSTROUTING", action, args); err != nil {
		return err
	}

	if proto == "sctp" {
		// Linux kernel v4.9 and below enables NETIF_F_SCTP_CRC for veth by
		// the following commit.
		// This introduces a problem when conbined with a physical NIC without
		// NETIF_F_SCTP_CRC. As for a workaround, here we add an iptables entry
		// to fill the checksum.
		//
		// https://github.com/torvalds/linux/commit/c80fafbbb59ef9924962f83aac85531039395b18
		args = []string{
			"-p", proto,
			"--sport", strconv.Itoa(destPort),
			"-j", "CHECKSUM",
			"--checksum-fill",
		}
		if err := ProgramRule(Mangle, "POSTROUTING", action, args); err != nil {
			return err
		}
	}

	return nil
}
619332
func (s *FirewallService) UpdatePortForwardingRule(p *UpdatePortForwardingRuleParams) (*UpdatePortForwardingRuleResponse, error) {
	resp, err := s.cs.newRequest("updatePortForwardingRule", p.toURLValues())
	if err != nil {
		return nil, err
	}

	var r UpdatePortForwardingRuleResponse
	if err := json.Unmarshal(resp, &r); err != nil {
		return nil, err
	}

	// If we have a async client, we need to wait for the async result
	if s.cs.async {
		b, err := s.cs.GetAsyncJobResult(r.JobID, s.cs.timeout)
		if err != nil {
			if err == AsyncTimeoutErr {
				return &r, err
			}
			return nil, err
		}

		b, err = getRawValue(b)
		if err != nil {
			return nil, err
		}

		b, err = convertFirewallServiceResponse(b)
		if err != nil {
			return nil, err
		}

		if err := json.Unmarshal(b, &r); err != nil {
			return nil, err
		}
	}

	return &r, nil
}