Эффективный способ использования csv.Reader () для "строки чана"

Question

У меня есть «строка чана», где каждая запись представляет собой строку журнала CSV, которую я хотел бы преобразовать в столбцы «[] строка», в настоящее время я (неэффективно) создаю csv.NewReader (strings.NewReader (i)) для каждого элемента, который выглядит намного больше работы, чем это должно быть на самом деле:

for i := range feederChan {
    r := csv.NewReader(strings.NewReader(i))
    a, err := r.Read()
    if err != nil {
         // log error...
         continue
    }
    // then do stuff with 'a'
    // ...
}

Итак, я бы очень хотел поделиться, если есть более эффективный способ сделать это, например, один раз создать csv.Reader, а затем каким-то образом передать ему содержимое чана (потоковое содержимое чана в то, что реализует 'io'. Интерфейс считывателя?).

Answer 1

Несмотря на то, что ответ «ThunderCat» был действительно полезен и оценен, я в итоге использовал io.Pipe () «как упоминалось в mh-cbon», который намного проще и выглядит более эффективным (объяснено ниже):

rp, wp := io.Pipe()
go func() {
    defer wp.Close()
    for i := range feederChan {
        fmt.Fprintln(wp, i)
    }
}()

r := csv.NewReader(rp)
for { // keep reading
    a, err := r.Read()
    if err == io.EOF {
        break
    }
    // do stuff with 'a'
    // ...
}

io.Pipe () является синхронным и должен быть достаточно эффективным: он передает данные от пишущего к читателю; Я скормил csv.NewReader () часть для чтения и создал процедуру, которая истощает запись чана в часть записи.

Большое спасибо.

РЕДАКТИРОВАТЬ: ThunderCat добавил подход io.Pipe к своему ответу (после того, как я опубликовал это, я думаю) ... его ответ гораздо более всеобъемлющий и был принят как таковой.

Answer 2

Используйте следующее для преобразования канала строк в читатель:

type chanReader struct {
    c   chan string
    buf string
}

func (r *chanReader) Read(p []byte) (int, error) {

    // Fill the buffer when we have no data to return to the caller
    if len(r.buf) == 0 {
        var ok bool
        r.buf, ok = <-r.c
        if !ok {
            // Return eof on channel closed
            return 0, io.EOF
        }
    }

    n := copy(p, r.buf)
    r.buf = r.buf[n:]
    return n, nil
}

Используйте это так:

r := csv.NewReader(&chanReader{c: feederChan})
for {
    a, err := r.Read()
    if err != nil {
        // handle error, break out of loop
    }
    // do something with a
}

Запустите его на детской площадке

Если приложение предполагает, что символы новой строки разделяют значения, полученные из канала, тогда добавьте новую строку к каждому полученному значению:

        ...
        var ok bool
        r.buf, ok = <-r.c
        if !ok {
            // Return eof on channel closed
            return 0, io.EOF
        }
        r.buf += "\n"
        ...

+= "\n" копирует строку.Если это не отвечает требованиям эффективности приложения, то введите новое поле для управления разделителями строк.

type chanReader struct {
    c chan string  // source of lines
    buf string     // the current line
    nl bool        // true if line separator is pending
}

func (r *chanReader) Read(p []byte) (int, error) {

    // Fill the buffer when we have no data to return to the caller
    if len(r.buf) == 0 && !r.nl {
        var ok bool
        r.buf, ok = <-r.c
        if !ok {
            // Return eof on channel closed
            return 0, io.EOF
        }
        r.nl = true
    }

    // Return data if we have it
    if len(r.buf) > 0 {
        n := copy(p, r.buf)
        r.buf = r.buf[n:]
        return n, nil
    }

    // No data, return the line separator
    n := copy(p, "\n")
    r.nl = n == 0
    return n, nil
}

Запустите его на игровой площадке .

Другой подход заключается виспользовать io.Pipe и goroutine для преобразования канала в io.Reader, как это предлагается в комментарии к вопросу.Первый проход в этом подходе:

var nl = []byte("\n")

func createChanReader(c chan string) io.Reader {
    r, w := io.Pipe()
    go func() {
        defer w.Close()
        for s := range c {
            io.WriteString(w, s)
            w.Write(nl)
            }
        }
    }()
    return r
}

Используйте его так:

r := csv.NewReader(createChanReader(feederChan))
for {
    a, err := r.Read()
    if err != nil {
        // handle error, break out of loop
    }
    // do something with a
}

Этот первый проход в решении io.Pipe пропускает программу, когда приложение выходит из цикла передчитая трубу в EOF.Приложение может выйти из строя раньше, потому что программа чтения CSV обнаружила синтаксическую ошибку, приложение запаниковало из-за ошибки программиста или по ряду других причин.

Чтобы исправить утечку программы, выйдите из программы записи при ошибке записи.и закройте трубочист, когда закончите чтение.

var nl = []byte("\n")

func createChanReader(c chan string) *io.PipeReader {
    r, w := io.Pipe()
    go func() {
        defer w.Close()
        for s := range c {
            if _, err := io.WriteString(w, s); err != nil {
                return
            }
            if _, err := w.Write(nl); err != nil {
                return
            }
        }
    }()
    return r
}

Используйте его так:

cr := createChanReader(feederChan)
defer cr.Close() // Required for goroutine cleanup
r := csv.NewReader(cr)
for {
    a, err := r.Read()
    if err != nil {
        // handle error, break out of loop
    }
    // do something with a
}

Запустите его на детской площадке .