Add concurrency support

For #18

start.go

@@ -1,11 +1,13 @@
 package main
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"os/signal"
 	"path/filepath"
 	"strconv"
+	"strings"
 	"sync"
 	"time"
 )
@@ -13,6 +15,7 @@ import (
 const shutdownGraceTime = 3 * time.Second
 
 var flagPort int
+var flagConcurrency string
 
 var processes = map[string]*Process{}
 var shutdown_mutex = new(sync.Mutex)
@@ -37,6 +40,35 @@ func init() {
 	cmdStart.Flag.StringVar(&flagProcfile, "f", "Procfile", "procfile")
 	cmdStart.Flag.StringVar(&flagEnv, "e", "", "env")
 	cmdStart.Flag.IntVar(&flagPort, "p", 5000, "port")
+	cmdStart.Flag.StringVar(&flagConcurrency, "c", "", "concurrency")
+}
+
+func parseConcurrency(value string) (map[string]int, error) {
+	concurrency := map[string]int{}
+	if strings.TrimSpace(value) == "" {
+		return concurrency, nil
+	}
+
+	parts := strings.Split(value, ",")
+	for _, part := range parts {
+		if !strings.Contains(part, "=") {
+			return concurrency, errors.New("Parsing concurency")
+		}
+
+		nameValue := strings.Split(part, "=")
+		n, v := strings.TrimSpace(nameValue[0]), strings.TrimSpace(nameValue[1])
+		if n == "" || v == "" {
+			return concurrency, errors.New("Parsing concurency")
+		}
+
+		numProcs, err := strconv.ParseInt(v, 10, 16)
+		if err != nil {
+			return concurrency, err
+		}
+
+		concurrency[n] = int(numProcs)
+	}
+	return concurrency, nil
 }
 
 func runStart(cmd *Command, args []string) {
@@ -52,6 +84,9 @@ func runStart(cmd *Command, args []string) {
 	env, err := ReadEnv(flagEnv)
 	handleError(err)
 
+	concurrency, err := parseConcurrency(flagConcurrency)
+	handleError(err)
+
 	of := NewOutletFactory()
 	of.Padding = pf.LongestProcessName()
 
@@ -77,26 +112,35 @@ func runStart(cmd *Command, args []string) {
 	}
 
 	for idx, proc := range pf.Entries {
-		if (singleton == "") || (singleton == proc.Name) {
-			shutdown_mutex.Lock()
-			wg.Add(1)
-			port := flagPort + (idx * 100)
-			ps := NewProcess(proc.Command, env)
-			processes[proc.Name] = ps
-			ps.Env["PORT"] = strconv.Itoa(port)
-			ps.Root = filepath.Dir(flagProcfile)
-			ps.Stdin = nil
-			ps.Stdout = of.CreateOutlet(proc.Name, idx, false)
-			ps.Stderr = of.CreateOutlet(proc.Name, idx, true)
-			ps.Start()
-			of.SystemOutput(fmt.Sprintf("starting %s on port %d", proc.Name, port))
-			go func(proc ProcfileEntry, ps *Process) {
-				ps.Wait()
-				wg.Done()
-				delete(processes, proc.Name)
-				ShutdownProcesses(of)
-			}(proc, ps)
-			shutdown_mutex.Unlock()
+		numProcs := 1
+		if value, ok := concurrency[proc.Name]; ok {
+			numProcs = value
+		}
+		for i := 0; i < numProcs; i++ {
+			if (singleton == "") || (singleton == proc.Name) {
+				shutdown_mutex.Lock()
+				wg.Add(1)
+				port := flagPort + (idx * 100)
+				ps := NewProcess(proc.Command, env)
+				procName := strings.Join([]string{
+					proc.Name,
+					strconv.FormatInt(int64(i+1), 10)}, ".")
+				processes[procName] = ps
+				ps.Env["PORT"] = strconv.Itoa(port)
+				ps.Root = filepath.Dir(flagProcfile)
+				ps.Stdin = nil
+				ps.Stdout = of.CreateOutlet(procName, idx, false)
+				ps.Stderr = of.CreateOutlet(procName, idx, true)
+				ps.Start()
+				of.SystemOutput(fmt.Sprintf("starting %s on port %d", procName, port))
+				go func(proc ProcfileEntry, ps *Process) {
+					ps.Wait()
+					wg.Done()
+					delete(processes, procName)
+					ShutdownProcesses(of)
+				}(proc, ps)
+				shutdown_mutex.Unlock()
+			}
 		}
 	}
 

start_test.go

@@ -0,0 +1,108 @@
+package main
+
+import "testing"
+
+func TestParseConcurrencyFlagEmpty(t *testing.T) {
+	c, err := parseConcurrency("")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(c) > 0 {
+		t.Fatal("expected no concurrency settings with ''")
+	}
+}
+
+func TestParseConcurrencyFlagSimle(t *testing.T) {
+	c, err := parseConcurrency("foo=2")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(c) != 1 {
+		t.Fatal("expected 1 concurrency settings with 'foo=2'")
+	}
+
+	if c["foo"] != 2 {
+		t.Fatal("expected concurrency settings of 2 with 'foo=2'")
+	}
+}
+
+func TestParseConcurrencyFlagMultiple(t *testing.T) {
+	c, err := parseConcurrency("foo=2,bar=3")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if len(c) != 2 {
+		t.Fatal("expected 1 concurrency settings with 'foo=2'")
+	}
+
+	if c["foo"] != 2 {
+		t.Fatal("expected concurrency settings of 2 with 'foo=2'")
+	}
+
+	if c["bar"] != 3 {
+		t.Fatal("expected concurrency settings of 3 with 'bar=3'")
+	}
+}
+
+func TestParseConcurrencyFlagNonInt(t *testing.T) {
+	_, err := parseConcurrency("foo=x")
+	if err == nil {
+		t.Fatal("foo=x should fail")
+	}
+}
+
+func TestParseConcurrencyFlagWhitespace(t *testing.T) {
+	c, err := parseConcurrency("foo   =   2, bar = 3")
+	if err != nil {
+		t.Fatalf("foo   =   2, bar = 4 should not fail:%s", err)
+	}
+
+	if len(c) != 2 {
+		t.Fatal("expected 1 concurrency settings with 'foo=2'")
+	}
+
+	if c["foo"] != 2 {
+		t.Fatal("expected concurrency settings of 2 with 'foo=2'")
+	}
+
+	if c["bar"] != 3 {
+		t.Fatal("expected concurrency settings of 3 with 'bar=3'")
+	}
+}
+
+func TestParseConcurrencyFlagMultipleEquals(t *testing.T) {
+	_, err := parseConcurrency("foo===2")
+	if err == nil {
+		t.Fatalf("foo===2 should fail: %s", err)
+	}
+}
+
+func TestParseConcurrencyFlagNoValue(t *testing.T) {
+	_, err := parseConcurrency("foo=")
+	if err == nil {
+		t.Fatalf("foo= should fail: %s", err)
+	}
+
+	_, err = parseConcurrency("=")
+	if err == nil {
+		t.Fatalf("= should fail: %s", err)
+	}
+
+	_, err = parseConcurrency("=1")
+	if err == nil {
+		t.Fatalf("= should fail: %s", err)
+	}
+
+	_, err = parseConcurrency(",")
+	if err == nil {
+		t.Fatalf(", should fail: %s", err)
+	}
+
+	_, err = parseConcurrency(",,,")
+	if err == nil {
+		t.Fatalf(",,, should fail: %s", err)
+	}
+
+}