add heartbeat channels

龙赣华

2019-06-03 5860c5fd3e6ae9f5412c12cbf16086e8585f3ef1

add heartbeat channels

 client.go

@@ -3,7 +3,6 @@
import (
    "errors"
    "log"
    "strings"

    "nanomsg.org/go-mangos"
    "nanomsg.org/go-mangos/protocol/respondent"
@@ -27,6 +26,8 @@
    // http://192.168.1.1:8080
    service string

    heartbeatmsg chan []byte

    ctx    context.Context
    cancel context.CancelFunc
    sock   mangos.Socket
@@ -41,7 +42,7 @@
    ctx  context.Context
    sock mangos.Socket

    changes chan []string
    changes chan []byte
}

func Client(urlServer string, service string) (*DiscoveryClient, error) {
@@ -76,20 +77,25 @@
    }

    client := &DiscoveryClient{
        urlServer:  urlServer,
        urlPubSub:  urlPubSub,
        service:    service,
        ctx:        ctx,
        cancel:     cancel,
        sock:       sock,
        subscriber: subscriber,
        urlServer:    urlServer,
        urlPubSub:    urlPubSub,
        service:      service,
        ctx:          ctx,
        cancel:       cancel,
        sock:         sock,
        heartbeatmsg: make(chan []byte),
        subscriber:   subscriber,
    }

    go client.run()
    return client, nil
}

func (d *DiscoveryClient) Peers() (chan []string, error) {
func (d *DiscoveryClient) HeartBeatMsg() chan []byte {
    return d.heartbeatmsg
}

func (d *DiscoveryClient) Peers() (chan []byte, error) {
    if d.subscriber == nil {
        return nil, errors.New("No subscribe url is provided to discover the Peers")
    }
@@ -102,21 +108,27 @@

func (d *DiscoveryClient) run() {
    var err error
    var msg []byte
    for {
        _, err = d.sock.Recv()

        if err != nil {
            log.Println("DiscoveryClient: Cannot receive the SURVEY", err.Error())
        } else {
        select {
        case <-d.ctx.Done():
            close(d.heartbeatmsg)
            return
        default:
            msg, err = d.sock.Recv()
            if err != nil {
                log.Println("DiscoveryClient: Cannot receive the SURVEY", err.Error())
                continue
            }
            err = d.sock.Send([]byte(d.service))
            if err != nil {
                log.Println("DiscoveryClient: Cannot send the SURVEY response", err.Error())
                continue
            }
            select {
            case <-d.ctx.Done():
                return

            case d.heartbeatmsg <- msg:
                log.Println("recv heartbeat msg. ", msg)
            default:
                err = d.sock.Send([]byte(d.service))
                if err != nil {
                    log.Println("DiscoveryClient: Cannot send the SURVEY response", err.Error())
                }
            }
        }
    }
@@ -147,14 +159,14 @@
        url:     url,
        ctx:     ctx,
        sock:    sock,
        changes: make(chan []string, 8),
        changes: make(chan []byte, 8),
    }

    go subscriber.run()
    return subscriber, nil
}

func (s *Subscriber) Changes() chan []string {
func (s *Subscriber) Changes() chan []byte {
    return s.changes
}

@@ -171,15 +183,8 @@
            msg, err = s.sock.Recv()
            if err != nil {
                log.Println("DiscoveryClient: Cannot SUBSCRIBE to the changes", err.Error())

            }

            // non-blocking send to the channel, discards changes if the channel is not ready
            select {
            case s.changes <- strings.Split(string(msg), "|"):
            default:
            }

            s.changes <- msg
        }
    }
}