Add support for setting maxMsgSize

pkg/qflow/qflow.go

@@ -138,11 +138,16 @@ func (h *Handler) HandleRequest(w http.ResponseWriter, req *http.Request) {
 func ListenAndServe(config *Config, addr string, dataDir string) {
 	var ep []Endpoint
 	var timeout = config.HTTP.Timeout
+	var maxMsgSize = config.Queue.MaxMessageSize
 
 	if timeout.Seconds() == 0.0 {
 		timeout = 10 * time.Second
 	}
 
+	if maxMsgSize == 0 {
+		maxMsgSize = 1024 * 1024 * 10 // 10mb
+	}
+
 	if _, err := os.Stat(dataDir); os.IsNotExist(err) {
 		log.Infof("creating data directory: %s", dataDir)
 		err = os.MkdirAll(dataDir, 0755)
@@ -162,15 +167,15 @@ func ListenAndServe(config *Config, addr string, dataDir string) {
 		}
 
 		log.Infof("registered (%s) with endpoints: [%s]", endpoint.Name, strings.Join(endpoint.Hosts, ","))
-		log.Infof("config options: (http timeout: %s)", timeout)
+		log.Infof("config options: (http timeout: %s, maxMsgSize: %d)", timeout, maxMsgSize)
 
 		writer := make(chan interface{})
 		c := durable.Channel(writer, &durable.Config{
 			Name:            endpoint.Name,
 			DataPath:        dataDir,
-			MaxBytesPerFile: 102400,
+			MaxBytesPerFile: 1024 * 1024 * 1024,
 			MinMsgSize:      0,
-			MaxMsgSize:      1000,
+			MaxMsgSize:      maxMsgSize,
 			SyncEvery:       10000,
 			SyncTimeout:     time.Second * 10,
 		})

pkg/qflow/structs.go

@@ -11,6 +11,9 @@ type Config struct {
 	HTTP struct {
 		Timeout time.Duration `yaml:"timeout"`
 	}
+	Queue struct {
+		MaxMessageSize int32 `yaml:"maxMsgSize"`
+	}
 
 	Endpoints []struct {
 		Name  string   `yaml:"name"`