add rate (#33)

* add rate

Author: Gsantomaggio

perfTest/cmd/commands.go

@@ -23,16 +23,20 @@ var (
 	streams           []string
 	maxLengthBytes    string
 	printStatsV       bool
+	rate              int
+	batchSize         int
 )
 
 func init() {
 	setupCli(rootCmd)
 }
 
 func setupCli(baseCmd *cobra.Command) {
+	batchSize = 100
 	baseCmd.PersistentFlags().StringVarP(&rabbitmqBrokerUrl, "uris", "u", streaming.LocalhostUriConnection, "Broker URL")
 	baseCmd.PersistentFlags().IntVarP(&producers, "producers", "p", 1, "Number of Producers")
 	baseCmd.PersistentFlags().IntVarP(&consumers, "consumers", "c", 1, "Number of Consumers")
+	baseCmd.PersistentFlags().IntVarP(&rate, "rate", "r", 0, "Limit publish rate")
 	baseCmd.PersistentFlags().BoolVarP(&preDeclared, "pre-declared", "d", false, "Pre created stream")
 	baseCmd.PersistentFlags().BoolVarP(&printStatsV, "print-stats", "n", true, "Print stats")
 	baseCmd.PersistentFlags().StringSliceVarP(&streams, "streams", "s", []string{uuid.New().String()}, "Stream names, create an UUID if not specified")

perfTest/cmd/silent.go

@@ -111,14 +111,18 @@ func startProducers() error {
 				return err
 			}
 			var arr []*amqp.Message
-			for z := 0; z < 100; z++ {
+			for z := 0; z < batchSize; z++ {
 
 				arr = append(arr, amqp.NewMessage([]byte(fmt.Sprintf("simul_%s", stream)  )))
 			}
 
 			go func(prod *streaming.Producer, messages []*amqp.Message) {
 				for {
-					//time.Sleep(1 * time.Millisecond)
+					if rate > 0 {
+						sleep := float64(batchSize) / float64(rate)
+						sleep = sleep * 1000
+						time.Sleep(time.Duration(sleep) * time.Millisecond)
+					}
 					atomic.AddInt32(&producerMessageCount, 100)
 					_, err = prod.BatchPublish(nil, arr)
 					if err != nil {

pkg/streaming/client_test.go

@@ -105,5 +105,19 @@ var _ = Describe("Streaming testClient", func() {
 			Expect(err).NotTo(HaveOccurred())
 		})
 
+		It("Create two times Stream precondition fail", func() {
+			err := testClient.StreamCreator().Stream(testStreamName).Create()
+			Expect(err).NotTo(HaveOccurred())
+			err = testClient.StreamCreator().Stream(testStreamName).
+				MaxLengthBytes(ByteCapacity{}.MB(100)).
+				Create()
+			Expect(err).To(HaveOccurred())
+			Expect(fmt.Sprintf("%s", err)).
+				To(ContainSubstring("Precondition Failed"))
+			err = testClient.DeleteStream(testStreamName)
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+
 	})
 })

pkg/streaming/constants.go

@@ -83,6 +83,8 @@ func LookErrorCode(errorCode uint16) string {
 		return "Code subscription id does not exist"
 	case ResponseCodePublisherDoesNotExist:
 		return "Code publisher does not exist"
+	case ResponseCodePreconditionFailed:
+		return "Code Precondition Failed"
 	default:
 		{
 			WARN("Error not handled %d", errorCode)