fix several command print issue

xsql/processors/xsql_processor.go

@@ -78,14 +78,14 @@ func (p *StreamProcessor) execCreateStream(stmt *xsql.StreamStmt, db *badger.DB)
 	if err != nil {
 		return "", err
 	}else{
-		return fmt.Sprintf("stream %s created", stmt.Name), nil
+		return fmt.Sprintf("Stream %s is created.\n", stmt.Name), nil
 	}
 }
 
 func (p *StreamProcessor) execShowStream(stmt *xsql.ShowStreamsStatement, db *badger.DB) ([]string,error) {
 	keys, err := common.DbKeys(db)
 	if len(keys) == 0 {
-		keys = append(keys, "no stream definition found")
+		keys = append(keys, "No stream definitions are found.\n")
 	}
 	return keys, err
 }
@@ -93,14 +93,14 @@ func (p *StreamProcessor) execShowStream(stmt *xsql.ShowStreamsStatement, db *ba
 func (p *StreamProcessor) execDescribeStream(stmt *xsql.DescribeStreamStatement, db *badger.DB) (string,error) {
 	s, err := common.DbGet(db, string(stmt.Name))
 	if err != nil {
-		return "", fmt.Errorf("stream %s not found", stmt.Name)
+		return "", fmt.Errorf("Stream %s is not found.\n", stmt.Name)
 	}
 
 	parser := xsql.NewParser(strings.NewReader(s))
 	stream, err := xsql.Language.Parse(parser)
 	streamStmt, ok := stream.(*xsql.StreamStmt)
 	if !ok{
-		return "", fmt.Errorf("error resolving the stream %s, the data in db may be corrupted", stmt.Name)
+		return "", fmt.Errorf("Error resolving the stream %s, the data in db may be corrupted.\n", stmt.Name)
 	}
 	var buff bytes.Buffer
 	buff.WriteString("Fields\n--------------------------------------------------------------------------------\n")
@@ -117,7 +117,7 @@ func (p *StreamProcessor) execDescribeStream(stmt *xsql.DescribeStreamStatement,
 func (p *StreamProcessor) execExplainStream(stmt *xsql.ExplainStreamStatement, db *badger.DB) (string,error) {
 	_, err := common.DbGet(db, string(stmt.Name))
 	if err != nil{
-		return "", fmt.Errorf("stream %s not found", stmt.Name)
+		return "", fmt.Errorf("Stream %s is not found.\n", stmt.Name)
 	}
 	return "TO BE SUPPORTED", nil
 }
@@ -127,7 +127,7 @@ func (p *StreamProcessor) execDropStream(stmt *xsql.DropStreamStatement, db *bad
 	if err != nil {
 		return "", err
 	}else{
-		return fmt.Sprintf("stream %s dropped", stmt.Name), nil
+		return fmt.Sprintf("Stream %s is dropped.\n", stmt.Name), nil
 	}
 }
 
@@ -141,7 +141,7 @@ func GetStream(db *badger.DB, name string) (stmt *xsql.StreamStmt, err error){
 	stream, err := xsql.Language.Parse(parser)
 	stmt, ok := stream.(*xsql.StreamStmt)
 	if !ok{
-		err = fmt.Errorf("error resolving the stream %s, the data in db may be corrupted", name)
+		err = fmt.Errorf("Error resolving the stream %s, the data in db may be corrupted.\n", name)
 	}
 	return
 }
@@ -172,7 +172,7 @@ func (p *RuleProcessor) ExecCreate(name, ruleJson string) (*api.Rule, error) {
 		common.DbClose(db)
 		return nil, err
 	}else{
-		log.Infof("rule %s created", name)
+		log.Infof("Rule %s is created.\n", name)
 		common.DbClose(db)
 	}
 	return rule, nil
@@ -186,7 +186,7 @@ func (p *RuleProcessor) GetRuleByName(name string) (*api.Rule, error) {
 	defer common.DbClose(db)
 	s, err := common.DbGet(db, string(name))
 	if err != nil {
-		return nil, fmt.Errorf("rule %s not found", name)
+		return nil, fmt.Errorf("Rule %s is not found.\n", name)
 	}
 	return p.getRuleByJson(name, s)
 }
@@ -194,18 +194,18 @@ func (p *RuleProcessor) GetRuleByName(name string) (*api.Rule, error) {
 func (p *RuleProcessor) getRuleByJson(name, ruleJson string) (*api.Rule, error) {
 	var rule api.Rule
 	if err := json.Unmarshal([]byte(ruleJson), &rule); err != nil {
-		return nil, fmt.Errorf("parse rule %s error : %s", ruleJson, err)
+		return nil, fmt.Errorf("Parse rule %s error : %s.\n", ruleJson, err)
 	}
 	rule.Id = name
 	//validation
 	if name == ""{
-		return nil, fmt.Errorf("missing rule id")
+		return nil, fmt.Errorf("Missing rule id.\n")
 	}
 	if rule.Sql == ""{
-		return nil, fmt.Errorf("missing rule sql")
+		return nil, fmt.Errorf("Missing rule SQL.\n")
 	}
 	if rule.Actions == nil || len(rule.Actions) == 0{
-		return nil, fmt.Errorf("missing rule actions")
+		return nil, fmt.Errorf("Missing rule actions.\n")
 	}
 	return &rule, nil
 }
@@ -218,17 +218,17 @@ func (p *RuleProcessor) ExecInitRule(rule *api.Rule) (*xstream.TopologyNew, erro
 			for name, action := range m {
 				switch name {
 				case "log":
-					log.Printf("Create log sink with %s", action)
+					log.Printf("Create log sink with %s.\n", action)
 					tp.AddSink(inputs, nodes.NewSinkNode("sink_log", sinks.NewLogSink()))
 				case "mqtt":
-					log.Printf("Create mqtt sink with %s", action)
+					log.Printf("Create mqtt sink with %s.\n", action)
 					if ms, err := sinks.NewMqttSink(action); err != nil{
 						return nil, err
 					}else{
 						tp.AddSink(inputs, nodes.NewSinkNode("sink_mqtt", ms))
 					}
 				default:
-					return nil, fmt.Errorf("unsupported action: %s", name)
+					return nil, fmt.Errorf("unsupported action: %s.\n", name)
 				}
 			}
 		}
@@ -260,7 +260,7 @@ func (p *RuleProcessor) ExecDesc(name string) (string, error) {
 	defer common.DbClose(db)
 	s, err := common.DbGet(db, string(name))
 	if err != nil {
-		return "", fmt.Errorf("rule %s not found", name)
+		return "", fmt.Errorf("Rule %s is not found.\n", name)
 	}
 	dst := &bytes.Buffer{}
 	if err := json.Indent(dst, []byte(s), "", "  "); err != nil {
@@ -276,7 +276,7 @@ func (p *RuleProcessor) ExecShow() (string, error) {
 		return "", err
 	}
 	if len(keys) == 0 {
-		keys = append(keys, "no rule definition found")
+		keys = append(keys, "No rule definitions are found.\n")
 	}
 	var result string
 	for _, c := range keys{
@@ -304,7 +304,7 @@ func (p *RuleProcessor) ExecDrop(name string) (string, error) {
 	if err != nil {
 		return "", err
 	}else{
-		return fmt.Sprintf("rule %s dropped", name), nil
+		return fmt.Sprintf("Rule %s is dropped.\n", name), nil
 	}
 }
 
@@ -321,7 +321,7 @@ func (p *RuleProcessor) createTopoWithSources(rule *api.Rule, sources []*nodes.S
 	if iet, ok := rule.Options["isEventTime"]; ok{
 		isEventTime, ok = iet.(bool)
 		if !ok{
-			return nil, nil, fmt.Errorf("invalid rule option isEventTime %v, bool type required", iet)
+			return nil, nil, fmt.Errorf("Invalid rule option isEventTime %v, bool type is required.\n", iet)
 		}
 	}
 	if isEventTime {
@@ -329,23 +329,23 @@ func (p *RuleProcessor) createTopoWithSources(rule *api.Rule, sources []*nodes.S
 			if fl, ok := l.(float64); ok{
 				lateTol = int64(fl)
 			}else{
-				return nil, nil, fmt.Errorf("invalid rule option lateTolerance %v, int type required", l)
+				return nil, nil, fmt.Errorf("Invalid rule option lateTolerance %v, int type is required.\n", l)
 			}
 		}
 	}
 	shouldCreateSource := sources == nil
 	parser := xsql.NewParser(strings.NewReader(sql))
 	if stmt, err := xsql.Language.Parse(parser); err != nil{
-		return nil, nil, fmt.Errorf("parse sql %s error: %s", sql , err)
+		return nil, nil, fmt.Errorf("Parse SQL %s error: %s.\n", sql , err)
 	}else {
 		if selectStmt, ok := stmt.(*xsql.SelectStatement); !ok {
-			return nil, nil, fmt.Errorf("sql %s is not a select statement", sql)
+			return nil, nil, fmt.Errorf("SQL %s is not a select statement.\n", sql)
 		} else {
 			tp := xstream.NewWithName(name)
 			var inputs []api.Emitter
 			streamsFromStmt := xsql.GetStreams(selectStmt)
 			if !shouldCreateSource && len(streamsFromStmt) != len(sources){
-				return nil, nil, fmt.Errorf("invalid parameter sources or streams, the length cannot match the statement, expect %d sources", len(streamsFromStmt))
+				return nil, nil, fmt.Errorf("Invalid parameter sources or streams, the length cannot match the statement, expect %d sources.\n", len(streamsFromStmt))
 			}
 			db, err := common.DbOpen(path.Join(p.badgerDir, "stream"))
 			if err != nil {

xstream/cli/main.go

@@ -21,7 +21,7 @@ type clientConf struct {
 
 var clientYaml = "client.yaml"
 
-func streamProcess(client *rpc.Client, args string) error {
+func streamProcess(client *rpc.Client, args string)  {
 	var reply string
 	if args == ""{
 		args = strings.Join(os.Args[1:], " ")
@@ -29,11 +29,9 @@ func streamProcess(client *rpc.Client, args string) error {
 	err := client.Call("Server.Stream", args, &reply)
 	if err != nil{
 		fmt.Println(err)
-		return err
 	}else{
 		fmt.Println(reply)
 	}
-	return nil
 }
 
 func main() {
@@ -65,7 +63,7 @@ func main() {
 		}
 	}
 
-	fmt.Printf("Connecting to %s:%d \n", config.Host, config.Port)
+	fmt.Printf("Connecting to %s:%d... \n", config.Host, config.Port)
 	// Create a TCP connection to localhost on port 1234
 	client, err := rpc.DialHTTP("tcp", fmt.Sprintf("%s:%d", config.Host, config.Port))
 	if err != nil {
@@ -139,19 +137,21 @@ func main() {
 						sfile := c.String("file")
 						if sfile != "" {
 							if _, err := os.Stat(c.String("file")); os.IsNotExist(err) {
-								fmt.Printf("The specified stream defintion file %s does not existed.", sfile)
+								fmt.Printf("The specified stream defintion file %s does not existed.\n", sfile)
 								return nil
 							}
-							fmt.Printf("Creating a new stream from file %s", sfile)
+							fmt.Printf("Creating a new stream from file %s.\n", sfile)
 							if stream, err := ioutil.ReadFile(sfile); err != nil {
-								fmt.Printf("Failed to read from stream definition file %s", sfile)
+								fmt.Printf("Failed to read from stream definition file %s.\n", sfile)
 								return nil
 							} else {
 								args := strings.Join([]string{"CREATE STREAM ", string(stream)}, " ")
-								return streamProcess(client, args)
+								streamProcess(client, args)
+								return nil
 							}
 						} else {
-							return streamProcess(client, "")
+							streamProcess(client, "")
+							return nil
 						}
 					},
 				},
@@ -169,12 +169,12 @@ func main() {
 						sfile := c.String("file")
 						if sfile != "" {
 							if _, err := os.Stat(c.String("file")); os.IsNotExist(err) {
-								fmt.Printf("The specified rule defenition file %s does not existed.", sfile)
+								fmt.Printf("The specified rule defenition file %s is not existed.\n", sfile)
 								return nil
 							}
-							fmt.Printf("Creating a new rule from file %s", sfile)
+							fmt.Printf("Creating a new rule from file %s.\n", sfile)
 							if rule, err := ioutil.ReadFile(sfile); err != nil {
-								fmt.Printf("Failed to read from rule definition file %s", sfile)
+								fmt.Printf("Failed to read from rule definition file %s.\n", sfile)
 								return nil
 							} else {
 								if len(c.Args()) != 1 {
@@ -224,7 +224,8 @@ func main() {
 					Usage: "describe stream $stream_name",
 					//Flags: nflag,
 					Action: func(c *cli.Context) error {
-						return streamProcess(client, "")
+						streamProcess(client, "")
+						return nil
 					},
 				},
 				{
@@ -259,7 +260,8 @@ func main() {
 					Usage: "drop stream $stream_name",
 					//Flags: nflag,
 					Action: func(c *cli.Context) error {
-						return streamProcess(client, "")
+						streamProcess(client, "")
+						return nil
 					},
 				},
 				{
@@ -295,7 +297,8 @@ func main() {
 					Name:  "streams",
 					Usage: "show streams",
 					Action: func(c *cli.Context) error {
-						return streamProcess(client, "")
+						streamProcess(client, "")
+						return nil
 					},
 				},
 				{

xstream/server/main.go

@@ -36,13 +36,13 @@ func (t *Server) CreateQuery(sql string, reply *string) error {
 	}
 	tp, err := processors.NewRuleProcessor(path.Dir(dataDir)).ExecQuery(QUERY_RULE_ID, sql)
 	if err != nil {
-		msg := fmt.Sprintf("failed to create query: %s.", err)
+		msg := fmt.Sprintf("Failed to create query: %s.", err)
 		log.Println(msg)
 		return fmt.Errorf(msg)
 	} else {
 		rs := &RuleState{Name: QUERY_RULE_ID, Topology: tp, Triggered: true}
 		registry[QUERY_RULE_ID] = rs
-		msg := fmt.Sprintf("query is submit successfully.")
+		msg := fmt.Sprintf("Query was submit successfully.")
 		log.Println(msg)
 		*reply = fmt.Sprintf(msg)
 	}
@@ -77,8 +77,7 @@ func (t *Server) GetQueryResult(qid string, reply *string) error {
 func (t *Server) Stream(stream string, reply *string) error{
 	content, err := processors.NewStreamProcessor(stream, path.Join(path.Dir(dataDir), "stream")).Exec()
 	if err != nil {
-		fmt.Printf("stream command error: %s\n", err)
-		return err
+		return fmt.Errorf("Stream command error: %s", err)
 	} else {
 		for _, c := range content{
 			*reply = *reply + fmt.Sprintln(c)
@@ -90,9 +89,9 @@ func (t *Server) Stream(stream string, reply *string) error{
 func (t *Server) CreateRule(rule *common.Rule, reply *string) error{
 	r, err := processor.ExecCreate(rule.Name, rule.Json)
 	if err != nil {
-		return fmt.Errorf("create rule error : %s\n", err)
+		return fmt.Errorf("Create rule error : %s.", err)
 	} else {
-		*reply = fmt.Sprintf("rule %s created", rule.Name)
+		*reply = fmt.Sprintf("Rule %s was created.", rule.Name)
 	}
 	//Start the rule
 	rs, err := t.createRuleState(r)
@@ -123,7 +122,7 @@ func (t *Server) createRuleState(rule *api.Rule) (*RuleState, error){
 func (t *Server) GetStatusRule(name string, reply *string) error{
 	if rs, ok := registry[name]; ok{
 		if !rs.Triggered {
-			*reply = "stopped: canceled manually"
+			*reply = "Stopped: canceled manually."
 			return nil
 		}
 		c := (*rs.Topology).GetContext()
@@ -131,19 +130,19 @@ func (t *Server) GetStatusRule(name string, reply *string) error{
 			err := c.Err()
 			switch err{
 			case nil:
-				*reply = "running"
+				*reply = "Running\n"
 			case context.Canceled:
-				*reply = "stopped: canceled by error"
+				*reply = "Stopped: canceled by error."
 			case context.DeadlineExceeded:
-				*reply = "stopped: deadline exceed"
+				*reply = "Stopped: deadline exceed."
 			default:
-				*reply = "stopped: unknown reason"
+				*reply = "Stopped: unknown reason."
 			}
 		}else{
-			*reply = "stopped: no context found"
+			*reply = "Stopped: no context found."
 		}
 	}else{
-		return fmt.Errorf("rule %s not found", name)
+		return fmt.Errorf("Rule %s is not found", name)
 	}
 	return nil
 }
@@ -165,7 +164,7 @@ func (t *Server) StartRule(name string, reply *string) error{
 	if err != nil{
 		return err
 	}
-	*reply = fmt.Sprintf("rule %s started", name)
+	*reply = fmt.Sprintf("Rule %s was started", name)
 	return nil
 }
 
@@ -186,9 +185,9 @@ func (t *Server) StopRule(name string, reply *string) error{
 	if rs, ok := registry[name]; ok{
 		(*rs.Topology).Cancel()
 		rs.Triggered = false
-		*reply = fmt.Sprintf("rule %s stopped", name)
+		*reply = fmt.Sprintf("Rule %s was stopped.", name)
 	}else{
-		*reply = fmt.Sprintf("rule %s not found", name)
+		*reply = fmt.Sprintf("Rule %s was not found.", name)
 	}
 	return nil
 }
@@ -202,14 +201,14 @@ func (t *Server) RestartRule(name string, reply *string) error{
 	if err != nil{
 		return err
 	}
-	*reply = fmt.Sprintf("rule %s restarted", name)
+	*reply = fmt.Sprintf("Rule %s was restarted.", name)
 	return nil
 }
 
 func (t *Server) DescRule(name string, reply *string) error{
 	r, err := processor.ExecDesc(name)
 	if err != nil {
-		return fmt.Errorf("desc rule error : %s\n", err)
+		return fmt.Errorf("Desc rule error : %s.", err)
 	} else {
 		*reply = r
 	}
@@ -219,7 +218,7 @@ func (t *Server) DescRule(name string, reply *string) error{
 func (t *Server) ShowRules(_ int, reply *string) error{
 	r, err := processor.ExecShow()
 	if err != nil {
-		return fmt.Errorf("show rule error : %s\n", err)
+		return fmt.Errorf("Show rule error : %s.", err)
 	} else {
 		*reply = r
 	}
@@ -229,7 +228,7 @@ func (t *Server) ShowRules(_ int, reply *string) error{
 func (t *Server) DropRule(name string, reply *string) error{
 	r, err := processor.ExecDrop(name)
 	if err != nil {
-		return fmt.Errorf("drop rule error : %s\n", err)
+		return fmt.Errorf("Drop rule error : %s.", err)
 	} else {
 		err := t.StopRule(name, reply)
 		if err != nil{
@@ -301,9 +300,9 @@ func main() {
 	if e != nil {
 		log.Fatal("Listen error: ", e)
 	}
-	msg := fmt.Sprintf("Serving Kuiper server on port %d", common.Config.Port)
+	msg := fmt.Sprintf("Serving kuiper on port %d... \n", common.Config.Port)
 	log.Info(msg)
-	fmt.Println(msg)
+	fmt.Printf(msg)
 	// Start accept incoming HTTP connections
 	err = http.Serve(listener, nil)
 	if err != nil {