2021-02-15 16:19:20 +01:00
|
|
|
package inputreader
|
|
|
|
|
|
|
|
import (
|
|
|
|
"bytes"
|
2021-02-16 10:44:00 +01:00
|
|
|
"encoding/base64"
|
2021-02-15 16:19:20 +01:00
|
|
|
"github.com/rjeczalik/notify"
|
|
|
|
"io"
|
|
|
|
"log"
|
|
|
|
"os"
|
|
|
|
)
|
|
|
|
|
2021-02-16 10:44:00 +01:00
|
|
|
// FileWatcherReader is an abstraction of a folder watcher
|
2021-02-15 16:19:20 +01:00
|
|
|
// and behaves like a reader
|
|
|
|
type FileWatcherReader struct {
|
|
|
|
// Folder to watch
|
2021-02-16 10:44:00 +01:00
|
|
|
folderfd *os.File
|
2021-02-15 16:19:20 +01:00
|
|
|
// Notify Channel
|
|
|
|
eic chan notify.EventInfo
|
2021-02-16 10:44:00 +01:00
|
|
|
// TearDown channel
|
|
|
|
exit chan string
|
2021-02-15 16:19:20 +01:00
|
|
|
// Current buffer
|
2021-02-16 11:30:12 +01:00
|
|
|
json bool
|
2021-02-17 15:05:17 +01:00
|
|
|
// Current file
|
|
|
|
curfile *os.File
|
|
|
|
// Current state Watching / Reading
|
|
|
|
watching bool
|
|
|
|
// Insert Separator
|
|
|
|
insertsep bool
|
2021-02-15 16:19:20 +01:00
|
|
|
}
|
|
|
|
|
2021-02-16 10:44:00 +01:00
|
|
|
// NewFileWatcherReader creates a new FileWatcherReader
|
2021-02-16 11:30:12 +01:00
|
|
|
// json specifies whether we now we handle json files
|
|
|
|
func NewFileWatcherReader(f *os.File, j bool) (*FileWatcherReader, error) {
|
2021-02-15 16:19:20 +01:00
|
|
|
r := &FileWatcherReader{
|
|
|
|
folderfd: f,
|
2021-02-17 15:05:17 +01:00
|
|
|
eic: make(chan notify.EventInfo, 4096),
|
|
|
|
json: j,
|
|
|
|
watching: true,
|
|
|
|
insertsep: false,
|
2021-02-15 16:19:20 +01:00
|
|
|
}
|
2021-02-16 19:10:36 +01:00
|
|
|
// go routine holding the watcher
|
|
|
|
go func() {
|
|
|
|
if err := notify.Watch("./...", r.eic, notify.InCloseWrite); err != nil {
|
|
|
|
log.Fatal(err)
|
|
|
|
}
|
|
|
|
defer notify.Stop(r.eic)
|
|
|
|
<-r.exit
|
|
|
|
}()
|
2021-02-15 16:19:20 +01:00
|
|
|
return r, nil
|
|
|
|
}
|
|
|
|
|
2021-02-16 11:30:12 +01:00
|
|
|
// Read waits for InCloseWrite file event uses a bytes reader to copy
|
|
|
|
// the resulting file encoded in b64 in p
|
2021-02-15 16:19:20 +01:00
|
|
|
func (fw *FileWatcherReader) Read(p []byte) (n int, err error) {
|
2021-02-17 15:05:17 +01:00
|
|
|
|
|
|
|
// Watching for new files to read
|
|
|
|
if fw.watching {
|
|
|
|
watchloop:
|
|
|
|
for {
|
|
|
|
select {
|
2021-02-16 10:44:00 +01:00
|
|
|
case ei := <-fw.eic:
|
2021-02-17 15:05:17 +01:00
|
|
|
//log.Println("Got event:", ei)
|
2021-02-16 10:44:00 +01:00
|
|
|
// New File, let's read its content
|
|
|
|
var err error
|
2021-02-17 15:05:17 +01:00
|
|
|
fw.curfile, err = os.Open(ei.Path())
|
2021-02-16 10:44:00 +01:00
|
|
|
if err != nil {
|
|
|
|
log.Fatal(err)
|
|
|
|
}
|
2021-02-17 15:05:17 +01:00
|
|
|
fw.watching = false
|
|
|
|
break watchloop
|
2021-02-16 10:44:00 +01:00
|
|
|
case <-fw.exit:
|
|
|
|
// Exiting
|
|
|
|
return 0, io.EOF
|
2021-02-17 15:05:17 +01:00
|
|
|
}
|
2021-02-16 10:44:00 +01:00
|
|
|
}
|
|
|
|
}
|
2021-02-17 15:05:17 +01:00
|
|
|
|
|
|
|
// Inserting separator
|
|
|
|
if fw.insertsep{
|
|
|
|
var buf []byte
|
|
|
|
buf = append(buf, "\n"...)
|
|
|
|
rreader := bytes.NewReader(buf)
|
|
|
|
n, err = rreader.Read(p)
|
|
|
|
fw.watching = true
|
|
|
|
fw.insertsep = false
|
|
|
|
log.Println("Inserting file seperator ")
|
|
|
|
fw.curfile.Close()
|
|
|
|
return n, err
|
|
|
|
}
|
|
|
|
|
|
|
|
// Reading
|
|
|
|
// if not json it could be anything so we encode it in b64
|
|
|
|
if !fw.json {
|
|
|
|
// base64 stream encoder
|
|
|
|
b64buffer := new(bytes.Buffer)
|
|
|
|
b64encoder := base64.NewEncoder(base64.StdEncoding, b64buffer)
|
2021-02-17 16:15:54 +01:00
|
|
|
// max buffer size is then 3072 = 4096/4*3
|
|
|
|
buf := make([]byte, 3072)
|
|
|
|
bytesread, err := fw.curfile.Read(buf)
|
2021-02-17 15:05:17 +01:00
|
|
|
// buf is the input
|
2021-02-17 16:15:54 +01:00
|
|
|
b64encoder.Write(buf[:bytesread])
|
2021-02-17 15:05:17 +01:00
|
|
|
// Close the encoder to flush partially written blocks
|
|
|
|
b64encoder.Close()
|
2021-02-17 16:15:54 +01:00
|
|
|
log.Println(b64buffer)
|
2021-02-17 15:05:17 +01:00
|
|
|
if err == io.EOF {
|
|
|
|
fw.insertsep = true
|
|
|
|
}
|
|
|
|
// Copy from b64buffer to p
|
2021-02-17 16:15:54 +01:00
|
|
|
n, err = b64buffer.Read(p[:len(b64buffer.Bytes())])
|
|
|
|
log.Println("len b64 buffer: ", len(b64buffer.Bytes()))
|
2021-02-17 15:05:17 +01:00
|
|
|
} else {
|
|
|
|
n, err = fw.curfile.Read(p)
|
|
|
|
if err == io.EOF {
|
|
|
|
fw.insertsep = true
|
|
|
|
}
|
|
|
|
}
|
2021-02-17 16:15:54 +01:00
|
|
|
log.Println("nread: ", n)
|
2021-02-17 15:05:17 +01:00
|
|
|
return n, err
|
2021-02-15 16:19:20 +01:00
|
|
|
}
|
|
|
|
|
2021-02-16 10:44:00 +01:00
|
|
|
// Teardown is called on error to stop the Reading loop if needed
|
|
|
|
func (rl *FileWatcherReader) Teardown() {
|
|
|
|
rl.exit <- "exit"
|
2021-02-17 15:05:17 +01:00
|
|
|
}
|