Fixed a number of compilation bugs.

This commit is contained in:
Nathan Osman 2016-04-25 01:01:01 -07:00
parent b62a889ab5
commit c0f04dc076
4 changed files with 57 additions and 42 deletions

View File

@ -13,16 +13,12 @@ type Entry struct {
} }
// LoadEntry loads an entry from disk. // LoadEntry loads an entry from disk.
func LoadEntry(filename string) (*Entry, error) { func (e *Entry) LoadEntry(filename string) error {
f, err := os.Open(filename) f, err := os.Open(filename)
if err != nil { if err != nil {
return nil, err return err
} }
e := &Entry{} return json.NewDecoder(f).Decode(e)
if err = json.NewDecoder(f).Decode(e); err != nil {
return nil, err
}
return e, nil
} }
// Save writes the entry to disk. // Save writes the entry to disk.

View File

@ -28,7 +28,7 @@ func NewReader(writer *Writer, jsonFilename, dataFilename string) *Reader {
status: StatusNone, status: StatusNone,
statusChanged: make(chan Status), statusChanged: make(chan Status),
} }
if r.writer { if r.writer != nil {
r.writer.Subscribe(r.statusChanged) r.writer.Subscribe(r.statusChanged)
} }
return r return r
@ -76,7 +76,7 @@ func (r *Reader) Read(p []byte) (n int, err error) {
err = errors.New("writer error") err = errors.New("writer error")
return return
case StatusDone: case StatusDone:
err = os.EOF err = io.EOF
return return
default: default:
for n < len(p) { for n < len(p) {
@ -84,7 +84,7 @@ func (r *Reader) Read(p []byte) (n int, err error) {
bytesRead, err = r.file.Read(p[n:]) bytesRead, err = r.file.Read(p[n:])
n += bytesRead n += bytesRead
if err != nil { if err != nil {
if err == os.EOF && r.writer != nil { if err == io.EOF && r.writer != nil {
err = nil err = nil
var watcher *fsnotify.Watcher var watcher *fsnotify.Watcher
watcher, err = fsnotify.NewWatcher() watcher, err = fsnotify.NewWatcher()

View File

@ -3,7 +3,6 @@ package main
import ( import (
"crypto/md5" "crypto/md5"
"fmt" "fmt"
"io"
"os" "os"
"path" "path"
"sync" "sync"
@ -13,27 +12,46 @@ import (
// avoid race conditions, adding and testing for entries in the cache are // avoid race conditions, adding and testing for entries in the cache are
// protected by a mutex. // protected by a mutex.
type Storage struct { type Storage struct {
Directory string directory string
mutex sync.Mutex writers map[string]*Writer
writerDone chan *Writer
mutex sync.Mutex
} }
// GetReader returns an io.Reader for the specified URL. If the file does not // NewStorage creates a new storage manager.
func NewStorage(directory string) *Storage {
return &Storage{
directory: directory,
writers: make(map[string]*Writer),
writerDone: make(chan *Writer),
}
}
// GetReader returns a *Reader for the specified URL. If the file does not
// exist, both a writer (for downloading the file) and a reader are created. // exist, both a writer (for downloading the file) and a reader are created.
func (s *Storage) GetReader(url string) (io.Reader, error) { func (s *Storage) GetReader(url string) (*Reader, error) {
var ( var (
hash = string(md5.Sum([]byte(url))) hash = string(md5.Sum([]byte(url)))
jsonFilename = path.Join(s.Directory, fmt.Sprintf("%s.json", hash)) jsonFilename = path.Join(s.directory, fmt.Sprintf("%s.json", hash))
dataFilename = path.Join(s.Directory, fmt.Sprintf("%s.data", hash)) dataFilename = path.Join(s.directory, fmt.Sprintf("%s.data", hash))
) )
s.mutex.Lock() s.mutex.Lock()
defer s.mutex.Unlock() defer s.mutex.Unlock()
i, err := os.Stat(jsonFilename) w, ok := s.writers[hash]
if err != nil { if ok {
if os.IsNotExist(err) { return NewReader(w, jsonFilename, dataFilename), nil
NewWriter(url, jsonFilename, dataFilename) } else {
_, err := os.Stat(jsonFilename)
if err != nil {
if os.IsNotExist(err) {
w = NewWriter(url, jsonFilename, dataFilename, s.writerDone)
s.writers[hash] = w
return NewReader(w, jsonFilename, dataFilename), nil
} else {
return nil, err
}
} else { } else {
return nil, err return NewReader(nil, jsonFilename, dataFilename), nil
} }
} }
return NewReader(url, jsonFilename, dataFilename)
} }

View File

@ -3,7 +3,6 @@ package main
import ( import (
"container/list" "container/list"
"io" "io"
"log"
"net/http" "net/http"
"os" "os"
"sync" "sync"
@ -27,7 +26,7 @@ type Writer struct {
status Status status Status
} }
func (w *Writer) sendStatus(statusChan chan Status, status Status) { func (w *Writer) sendStatus(statusChan chan<- Status, status Status) {
statusChan <- status statusChan <- status
} }
@ -36,17 +35,16 @@ func (w *Writer) setStatus(status Status) {
defer w.mutex.Unlock() defer w.mutex.Unlock()
w.status = status w.status = status
for e := w.channels.Front(); e != nil; e = e.Next() { for e := w.channels.Front(); e != nil; e = e.Next() {
go w.sendStatus(e.Value.(chan Status), status) go w.sendStatus(e.Value.(chan<- Status), status)
} }
} }
// Create a new writer for the given URL. Status information is passed along to // NewWriter creates a new writer for the given URL. Status information is
// subscribed channels. The done channel is used to notify the storage system // passed along to subscribed channels. The done channel is used to notify the
// that the download is complete (either success or an error). // storage system that the download is complete (either success or an error).
func NewWriter(url, jsonFilename, dataFilename string, done chan<- *Writer) *Writer { func NewWriter(url, jsonFilename, dataFilename string, done chan<- *Writer) *Writer {
w := &Writer{ w := &Writer{
channels: make([]chan Status), channels: list.New(),
status: StatusNone,
} }
go func() { go func() {
r, err := http.Get(url) r, err := http.Get(url)
@ -55,8 +53,9 @@ func NewWriter(url, jsonFilename, dataFilename string, done chan<- *Writer) *Wri
} }
defer r.Body.Close() defer r.Body.Close()
e := &Entry{ e := &Entry{
URL: url, URL: url,
ContentType: r.Header.Get("Content-Type"), ContentLength: r.ContentLength,
ContentType: r.Header.Get("Content-Type"),
} }
if err = e.Save(jsonFilename); err != nil { if err = e.Save(jsonFilename); err != nil {
goto error goto error
@ -82,21 +81,23 @@ func NewWriter(url, jsonFilename, dataFilename string, done chan<- *Writer) *Wri
// Subscribe adds a channel to the list to be notified when the writer's status // Subscribe adds a channel to the list to be notified when the writer's status
// changes. The channel will also immediately receive the current status. // changes. The channel will also immediately receive the current status.
func (w *Writer) Subscribe(statusChan chan Status) { func (w *Writer) Subscribe(statusChan chan<- Status) {
w.mutex.Lock() w.mutex.Lock()
defer w.mutex.Unlock() defer w.mutex.Unlock()
w.channels = append(w.channels, c) w.channels.PushBack(statusChan)
w.sendStatus(c, w.status)
// TODO: is "go" necessary here?
go w.sendStatus(statusChan, w.status)
} }
// Unsubscribe a channel from the list to be notified. This may occur when a // Unsubscribe removes a channel from the list to be notified. This may occur
// client cancels a request, for example. // when a client cancels a request, for example.
func (w *Writer) Unsubscribe(statusChan chan Status) { func (w *Writer) Unsubscribe(statusChan chan<- Status) {
w.mutex.Lock() w.mutex.Lock()
defer w.mutex.Unlock() defer w.mutex.Unlock()
for i, c := range w.channels { for e := w.channels.Front(); e != nil; e = e.Next() {
if c == statusChan { if e.Value.(chan<- Status) == statusChan {
w.channels = append(w.channels[:i], w.channels[i+1:]...) w.channels.Remove(e)
} }
} }
} }