mirror of
https://github.com/minio/minio.git
synced 2025-03-30 09:13:41 -04:00
Split's SplitStream now creates and returns a read only channel.
This commit is contained in:
parent
d73c57cff9
commit
c02fa65b0f
@ -55,7 +55,13 @@ type JoinMessage struct {
|
|||||||
// for chunk := range channel {
|
// for chunk := range channel {
|
||||||
// log.Println(chunk.Data)
|
// log.Println(chunk.Data)
|
||||||
// }
|
// }
|
||||||
func SplitStream(reader io.Reader, chunkSize uint64, ch chan SplitMessage) {
|
func SplitStream(reader io.Reader, chunkSize uint64) <-chan SplitMessage {
|
||||||
|
ch := make(chan SplitMessage)
|
||||||
|
go splitStreamGoRoutine(reader, chunkSize, ch)
|
||||||
|
return ch
|
||||||
|
}
|
||||||
|
|
||||||
|
func splitStreamGoRoutine(reader io.Reader, chunkSize uint64, ch chan SplitMessage) {
|
||||||
// we read until EOF or another error
|
// we read until EOF or another error
|
||||||
var readError error
|
var readError error
|
||||||
|
|
||||||
@ -199,8 +205,7 @@ func SplitFilesWithPrefix(filename string, chunkstr string, outputPrefix string)
|
|||||||
}
|
}
|
||||||
|
|
||||||
// start stream splitting goroutine
|
// start stream splitting goroutine
|
||||||
ch := make(chan SplitMessage)
|
ch := SplitStream(file, chunkSize)
|
||||||
go SplitStream(file, chunkSize, ch)
|
|
||||||
|
|
||||||
// used to write each chunk out as a separate file. {{outputPrefix}}.{{i}}
|
// used to write each chunk out as a separate file. {{outputPrefix}}.{{i}}
|
||||||
i := 0
|
i := 0
|
||||||
|
@ -38,9 +38,8 @@ func (s *MySuite) TestSplitStream(c *C) {
|
|||||||
bytesWriter.Write([]byte(strconv.Itoa(i)))
|
bytesWriter.Write([]byte(strconv.Itoa(i)))
|
||||||
}
|
}
|
||||||
bytesWriter.Flush()
|
bytesWriter.Flush()
|
||||||
ch := make(chan SplitMessage)
|
|
||||||
reader := bytes.NewReader(bytesBuffer.Bytes())
|
reader := bytes.NewReader(bytesBuffer.Bytes())
|
||||||
go SplitStream(reader, 25, ch)
|
ch := SplitStream(reader, 25)
|
||||||
var resultsBuffer bytes.Buffer
|
var resultsBuffer bytes.Buffer
|
||||||
resultsWriter := bufio.NewWriter(&resultsBuffer)
|
resultsWriter := bufio.NewWriter(&resultsBuffer)
|
||||||
for chunk := range ch {
|
for chunk := range ch {
|
||||||
|
@ -136,8 +136,7 @@ func (eStorage *encodedStorage) List(objectPath string) ([]storage.ObjectDescrip
|
|||||||
|
|
||||||
func (eStorage *encodedStorage) Put(objectPath string, object io.Reader) error {
|
func (eStorage *encodedStorage) Put(objectPath string, object io.Reader) error {
|
||||||
// split
|
// split
|
||||||
chunks := make(chan split.SplitMessage)
|
chunks := split.SplitStream(object, eStorage.BlockSize)
|
||||||
go split.SplitStream(object, eStorage.BlockSize, chunks)
|
|
||||||
|
|
||||||
// for each chunk
|
// for each chunk
|
||||||
encoderParameters, err := erasure.ParseEncoderParams(eStorage.K, eStorage.M, erasure.CAUCHY)
|
encoderParameters, err := erasure.ParseEncoderParams(eStorage.K, eStorage.M, erasure.CAUCHY)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user