1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
// Package to write WARC files.
package warc
import (
"bufio"
"fmt"
"io"
"time"
"compress/gzip"
"github.com/pborman/uuid"
)
var (
warcTimeFmt = "2006-01-02T15:04:05Z"
warcVersion = "WARC/1.0"
warcContentTypes = map[string]string{
"warcinfo": "application/warc-fields",
"response": "application/http; msgtype=response",
"request": "application/http; msgtype=request",
"metadata": "application/warc-fields",
}
)
// Header for a WARC record. Header field names are case-sensitive.
type Header map[string]string
// Set a header to the specified value. Multiple values are not
// supported.
func (h Header) Set(key, value string) {
h[key] = value
// Keep Content-Type in sync with WARC-Type.
if key == "WARC-Type" {
if ct, ok := warcContentTypes[value]; ok {
h["Content-Type"] = ct
} else {
h["Content-Type"] = "application/octet-stream"
}
}
}
// Get the value of a header. If not found, returns an empty string.
func (h Header) Get(key string) string {
return h[key]
}
// Encode the header to a Writer.
func (h Header) Encode(w io.Writer) error {
if _, err := fmt.Fprintf(w, "%s\r\n", warcVersion); err != nil {
return err
}
for hdr, value := range h {
if _, err := fmt.Fprintf(w, "%s: %s\r\n", hdr, value); err != nil {
return err
}
}
_, err := fmt.Fprintf(w, "\r\n")
return err
}
// NewHeader returns a Header with its own unique ID and the
// current timestamp.
func NewHeader() Header {
h := make(Header)
h.Set("WARC-Record-ID", fmt.Sprintf("<%s>", uuid.NewUUID().URN()))
h.Set("WARC-Date", time.Now().UTC().Format(warcTimeFmt))
h.Set("Content-Type", "application/octet-stream")
return h
}
// Writer can write records to a file in WARC format. It is safe
// for concurrent access, since writes are serialized internally.
type Writer struct {
writer io.WriteCloser
bufwriter *bufio.Writer
gzwriter *gzip.Writer
lockCh chan bool
}
type recordWriter struct {
io.Writer
lockCh chan bool
}
func (rw *recordWriter) Close() error {
// Add the end-of-record marker.
_, err := fmt.Fprintf(rw, "\r\n\r\n")
<-rw.lockCh
return err
}
// NewRecord starts a new WARC record with the provided header. The
// caller must call Close on the returned writer before creating the
// next record. Note that this function may block until that condition
// is satisfied. If this function returns an error, the state of the
// Writer is invalid and it should no longer be used.
func (w *Writer) NewRecord(hdr Header) (io.WriteCloser, error) {
w.lockCh <- true
if w.gzwriter != nil {
w.gzwriter.Close() // nolint
}
var err error
w.gzwriter, err = gzip.NewWriterLevel(w.bufwriter, gzip.BestCompression)
if err != nil {
return nil, err
}
w.gzwriter.Header.Name = hdr.Get("WARC-Record-ID")
if err = hdr.Encode(w.gzwriter); err != nil {
return nil, err
}
return &recordWriter{Writer: w.gzwriter, lockCh: w.lockCh}, nil
}
// Close the WARC writer and flush all buffers. This will also call
// Close on the wrapped io.WriteCloser object.
func (w *Writer) Close() error {
if err := w.gzwriter.Close(); err != nil {
return err
}
if err := w.bufwriter.Flush(); err != nil {
return err
}
return w.writer.Close()
}
// NewWriter initializes a new Writer and returns it.
func NewWriter(w io.WriteCloser) *Writer {
return &Writer{
writer: w,
bufwriter: bufio.NewWriter(w),
// Buffering is important here since we're using this
// channel as a semaphore.
lockCh: make(chan bool, 1),
}
}
|
