Merge pull request #158 from fkautz/pr_out_adding_initial_split_file_command

cmd/split-file/split-file-options.go

@@ -0,0 +1,80 @@
+/*
+ * Mini Object Storage, (C) 2014 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"io"
+	"log"
+	"os"
+	"path"
+
+	"github.com/codegangsta/cli"
+	"github.com/minio-io/minio/pkg/split"
+	"github.com/minio-io/minio/pkg/strbyteconv"
+)
+
+var Options = []cli.Command{
+	Split,
+	Merge,
+}
+
+var Split = cli.Command{
+	Name:        "split",
+	Usage:       "Describes how large each split should be",
+	Description: "",
+	Action:      doFileSplit,
+	Flags: []cli.Flag{
+		cli.StringFlag{
+			Name:  "size,s",
+			Value: "2M",
+			Usage: "",
+		},
+	},
+}
+
+var Merge = cli.Command{
+	Name:        "merge",
+	Usage:       "Describes how large each split should be",
+	Description: "",
+	Action:      doFileMerge,
+}
+
+func doFileSplit(c *cli.Context) {
+	chunkSize, err := strbyteconv.StringToBytes(c.String("size"))
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = split.SplitFileWithPrefix(c.Args().Get(0), chunkSize, c.Args().Get(1))
+	if err != nil {
+		// TODO cleanup?
+		log.Fatal(err)
+	}
+}
+
+func doFileMerge(c *cli.Context) {
+	prefix := c.Args().Get(0)
+	output := c.Args().Get(1)
+	prefix = path.Clean(prefix)
+	log.Println(path.Dir(prefix), path.Base(prefix))
+	reader := split.JoinFiles(path.Dir(prefix), path.Base(prefix))
+	file, err := os.OpenFile(output, os.O_WRONLY|os.O_CREATE, 0600)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer file.Close()
+	io.Copy(file, reader)
+}

cmd/split-file/split-file.go

@@ -0,0 +1,32 @@
+/*
+ * Mini Object Storage, (C) 2014 Minio, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import (
+	"os"
+
+	"github.com/codegangsta/cli"
+)
+
+func main() {
+	app := cli.NewApp()
+	app.Name = "split-file"
+	app.Usage = ""
+	app.Commands = Options
+	app.Author = "Minio"
+	app.Run(os.Args)
+}

cmd/split-file/split-file.md

@@ -0,0 +1,14 @@
+
+% MINIO(1) Minio Manual
+% Minio community
+% January 2015
+# NAME
+split-file - 
+
+# SYNOPSIS
+
+# DESCRIPTION
+
+# EXAMPLES
+
+# AUTHORS

pkg/split/split.go

@@ -22,7 +22,6 @@ import (
 	"bufio"
 	"bytes"
 	"errors"
-	"github.com/minio-io/minio/pkg/strbyteconv"
 	"io"
 	"io/ioutil"
 	"os"
@@ -101,93 +100,48 @@ func splitStreamGoRoutine(reader io.Reader, chunkSize uint64, ch chan SplitMessa
 	close(ch)
 }
 
-func JoinStream(dirname string, inputPrefix string) <-chan JoinMessage {
-	ch := make(chan JoinMessage)
-	go joinStreamGoRoutine(dirname, inputPrefix, ch)
-	return ch
-}
-
-func joinStreamGoRoutine(dirname string, inputPrefix string, ch chan JoinMessage) {
-	var readError error
-
-	var bytesBuffer bytes.Buffer
-	bytesWriter := bufio.NewWriter(&bytesBuffer)
-	// read a full directory
+func JoinFiles(dirname string, inputPrefix string) io.Reader {
+	reader, writer := io.Pipe()
 	fileInfos, readError := ioutil.ReadDir(dirname)
 	if readError != nil {
-		ch <- JoinMessage{nil, 0, readError}
+		writer.CloseWithError(readError)
 	}
 
 	var newfileInfos []os.FileInfo
 	for _, fi := range fileInfos {
 		if strings.Contains(fi.Name(), inputPrefix) == true {
 			newfileInfos = append(newfileInfos, fi)
-			continue
 		}
 	}
 
 	if len(newfileInfos) == 0 {
-		ch <- JoinMessage{nil, 0, errors.New("no files found for given prefix")}
-	}
-
-	for i := range newfileInfos {
-		slice, err := ioutil.ReadFile(newfileInfos[i].Name())
-		if err != nil {
-			ch <- JoinMessage{nil, 0, err}
-		}
-		bytesWriter.Write(slice)
-		bytesWriter.Flush()
-		if bytesBuffer.Len() != 0 {
-			ch <- JoinMessage{&bytesBuffer, newfileInfos[i].Size(), nil}
-		}
+		writer.CloseWithError(errors.New("no files found for given prefix"))
 	}
 
-	// close the channel, signaling the channel reader that the stream is complete
-	close(ch)
+	go joinFilesGoRoutine(newfileInfos, writer)
+	return reader
 }
 
-func JoinFilesWithPrefix(dirname string, inputPrefix string, outputFile string) error {
-	if dirname == "" {
-		return errors.New("Invalid directory")
-	}
-
-	if inputPrefix == "" {
-		return errors.New("Invalid argument inputPrefix cannot be empty string")
-	}
-
-	if outputFile == "" {
-		return errors.New("Invalid output file")
-	}
-
-	ch := JoinStream(dirname, inputPrefix)
-
-	var multiReaders []io.Reader
-	var aggregatedLength int64
-	for output := range ch {
-		if output.Err != nil {
-			return output.Err
+func joinFilesGoRoutine(fileInfos []os.FileInfo, writer *io.PipeWriter) {
+	for _, fileInfo := range fileInfos {
+		file, err := os.Open(fileInfo.Name())
+		defer file.Close()
+		for err != nil {
+			writer.CloseWithError(err)
+			return
+		}
+		_, err = io.Copy(writer, file)
+		if err != nil {
+			writer.CloseWithError(err)
+			return
 		}
-		multiReaders = append(multiReaders, output.Reader)
-		aggregatedLength += output.Length
-	}
-
-	newReader := io.MultiReader(multiReaders...)
-	aggregatedBytes := make([]byte, aggregatedLength)
-	_, err := newReader.Read(aggregatedBytes)
-	if err != nil {
-		return err
-	}
-
-	err = ioutil.WriteFile(outputFile, aggregatedBytes, 0600)
-	if err != nil {
-		return err
 	}
-	return nil
+	writer.Close()
 }
 
 // Takes a file and splits it into chunks with size chunkSize. The output
 // filename is given with outputPrefix.
-func SplitFilesWithPrefix(filename string, chunkstr string, outputPrefix string) error {
+func SplitFileWithPrefix(filename string, chunkSize uint64, outputPrefix string) error {
 	// open file
 	file, err := os.Open(filename)
 	defer file.Close()
@@ -199,11 +153,6 @@ func SplitFilesWithPrefix(filename string, chunkstr string, outputPrefix string)
 		return errors.New("Invalid argument outputPrefix cannot be empty string")
 	}
 
-	chunkSize, err := strbyteconv.StringToBytes(chunkstr)
-	if err != nil {
-		return err
-	}
-
 	// start stream splitting goroutine
 	ch := SplitStream(file, chunkSize)
 

pkg/split/split_test.go

@@ -19,6 +19,8 @@ package split
 import (
 	"bufio"
 	"bytes"
+	"io"
+	"os"
 	"strconv"
 	"testing"
 
@@ -50,13 +52,17 @@ func (s *MySuite) TestSplitStream(c *C) {
 }
 
 func (s *MySuite) TestFileSplitJoin(c *C) {
-	err := SplitFilesWithPrefix("TESTFILE", "1KB", "TESTPREFIX")
+	err := SplitFileWithPrefix("TESTFILE", 1024, "TESTPREFIX")
 	c.Assert(err, IsNil)
-	err = SplitFilesWithPrefix("TESTFILE", "1KB", "")
+	err = SplitFileWithPrefix("TESTFILE", 1024, "")
 	c.Assert(err, Not(IsNil))
 
-	err = JoinFilesWithPrefix(".", "TESTPREFIX", "")
+	devnull, err := os.OpenFile(os.DevNull, 2, os.ModeAppend)
+	defer devnull.Close()
+	reader := JoinFiles(".", "ERROR")
+	_, err = io.Copy(devnull, reader)
 	c.Assert(err, Not(IsNil))
-	err = JoinFilesWithPrefix(".", "TESTPREFIX", "NEWFILE")
+	reader = JoinFiles(".", "TESTPREFIX")
+	_, err = io.Copy(devnull, reader)
 	c.Assert(err, IsNil)
 }