Нужна помощь в понимании того, как работает libchan

Question

Я пытаюсь использовать библиотеку libchan для отправки сообщений между машинами с использованием транспорта, аналогичного каналу go.

Исходя из того, что я понял, грубая идея такова:

1. У вас есть клиент SPDY, который отправляет сериализованный объект команды по адресу через tcp.Этот объект команды содержит канал libchan с именем Pipe, по которому отправляется ответ.
2. Когда сервер получает входящее соединение, он ожидает объект команды.Когда он получает его, он отправляет ответ через Pipe, содержащийся в объекте.

Вот моя точка замешательства.Чтобы канал сохранялся между двумя машинами, им пришлось бы делиться памятью или, по крайней мере, делиться абстракцией, которая связывает их две.Из моего рокового отношения к базе данных libchan я понятия не имею, как это могло быть возможно.

Вот фрагмент из примера в репозитории:

// client

    receiver, remoteSender := libchan.Pipe()
    command := &RemoteCommand{
        Cmd:        os.Args[1],
        Args:       os.Args[2:],
        Stdin:      os.Stdin,
        Stdout:     os.Stdout,
        Stderr:     os.Stderr,
        StatusChan: remoteSender,
    }

    err = sender.Send(command)
    if err != nil {
        log.Fatal(err)
    }
    err = receiver.Receive(response)
    if err != nil {
        log.Fatal(err)
    }

    os.Exit(response.Status)

и сервера:

// server
t := spdy.NewTransport(p)

        go func() {
            for {
                receiver, err := t.WaitReceiveChannel()
                if err != nil {
                    log.Print("receiver error")
                    log.Print(err)
                    break
                }
                log.Print("about to spawn receive proc")
                go func() {
                    for {
                        command := &RemoteReceivedCommand{}
                        err := receiver.Receive(command)
                        returnResult := &CommandResponse{}
                        if res != nil {
                            if exiterr, ok := res.(*exec.ExitError); ok {
                                returnResult.Status = exiterr.Sys(). 
                              (syscall.WaitStatus).ExitStatus()
                            } else {
                                log.Print("res")
                                log.Print(res)
                                returnResult.Status = 10
                            }
                        }
                        err = command.StatusChan.Send(returnResult)

Точка, которую я пытаюсь отточить, находится здесь:

libchan.Pipe()

Согласно источнику, это возвращает канал.Одна ссылка хранится на клиенте, а другая отправляется на сервер.Этот канал затем используется для передачи значений от последнего к первому.Как это на самом деле работает на практике?

Полный код для клиента и сервера

Answer 1

Во-первых, хорошо знать, что все, что выполняет Pipe (), - это создание канала и возврат пары отправитель / получатель в памяти.

С inmem.go:

// Pipe returns an inmemory Sender/Receiver pair.
func Pipe() (Receiver, Sender) {
    c := make(chan interface{})
    return pReceiver(c), pSender(c)
}

Тогда вы можете посмотреть в inmem_test.go простой сквозной пример.

Эта структура является эквивалентом RemoteCommand из демоверсии.

type InMemMessage struct {
    Data   string
    Stream io.ReadWriteCloser
    Ret    Sender
}

В TestInmemRetPipe() создаются простой клиент и сервер.

Клиент создает локальную пару отправитель / получатель, используя Pipe (), а сервер просто использует libchan.Sender интерфейс в InMemMessage структура.

Обратите внимание, что клиент и сервер являются функциями, которые получают отправителя или получателя в качестве аргумента соответственно Подробнее об этом в следующем фрагменте кода.

func TestInmemRetPipe(t *testing.T) {
    client := func(t *testing.T, w Sender) {
        ret, retPipe := Pipe()
        message := &InMemMessage{Data: "hello", Ret: retPipe}

        err := w.Send(message)
        if err != nil {
            t.Fatal(err)
        }
        msg := &InMemMessage{}
        err = ret.Receive(msg)
        if err != nil {
            t.Fatal(err)
        }

        if msg.Data != "this better not crash" {
            t.Fatalf("%#v", msg)
        }

    }
    server := func(t *testing.T, r Receiver) {
        msg := &InMemMessage{}
        err := r.Receive(msg)
        if err != nil {
            t.Fatal(err)
        }

        if msg.Data != "hello" {
            t.Fatalf("Wrong message:\n\tExpected: %s\n\tActual: %s", "hello", msg.Data)
        }
        if msg.Ret == nil {
            t.Fatal("Message Ret is nil")
        }

        message := &InMemMessage{Data: "this better not crash"}
        if err := msg.Ret.Send(message); err != nil {
            t.Fatal(err)
        }
    }
    SpawnPipeTestRoutines(t, client, server)

}

SpawnPipeTestRoutines() выполняет функции клиента и сервера. В этой функции другой воздух отправителя / получателя создается с помощью Pipe().

В демонстрационном приложении функция, выполняемая здесь Pipe() (т. Е. Облегчающая связь между экземплярами клиента и сервера), вместо этого обрабатывается через сетевые коммуникации.

func SpawnPipeTestRoutines(t *testing.T, s SendTestRoutine, r ReceiveTestRoutine) {
    end1 := make(chan bool)
    end2 := make(chan bool)

    receiver, sender := Pipe()

    go func() {
        defer close(end1)
        s(t, sender)
        err := sender.Close()
        if err != nil {
            t.Fatalf("Error closing sender: %s", err)
        }
    }()

    go func() {
        defer close(end2)
        r(t, receiver)
    }()
    ...

В демонстрационном приложении связь облегчается за счет вызовов Transport.NewSendChannel () на клиенте и Transport.WaitReceiveChannel () , которые возвращают libchan.Sender и libchan.Receiver соответственно. Эти экземпляры libchan обрабатывают "канал" через сеть.

С client.go:

sender, err := transport.NewSendChannel()
...
err = sender.Send(command)

С server.go:

receiver, err := t.WaitReceiveChannel()
...
err := receiver.Receive(command)

В обоих случаях предварительная конфигурация транспорта выполняется заранее (то есть привязка к сокетам, использование TLS и т. Д.).

Вероятно, также стоит отметить, что используемая библиотека spdy является частью дистрибутива libchan , следовательно, она предоставляет примитивы libchan.