demo_test/test.go

// Copyright 2018 Open Networking Foundation
//
// 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 (
        "fmt"
        "net"
        "bufio"
        "os"
        "os/signal"
        "os/exec"
        "github.com/Shopify/sarama"
        "google.golang.org/grpc"
        "golang.org/x/net/context"
        importer "./proto"
        log "github.com/Sirupsen/logrus"
	"bytes"
	"strings"
        "net/http"
        "crypto/tls"
        "strconv"
)

var REDFISH_ROOT		= "/redfish/v1"
var CONTENT_TYPE		= "application/json"
var EVENTS_MAP = map[string]string{
"add":"ResourceAdded",
"rm":"ResourceRemoved",
"alert":"Alert",
"update":"Update"}

var default_address string	= "localhost:31085"
var default_port    string	= "8888"
var default_vendor  string	= "edgecore"
var default_freq    uint64	= 180
var attach_device_ip string	= ""
var importerTopic		= "importer"
var DataConsumer sarama.Consumer

var cc	   importer.DeviceManagementClient
var ctx	   context.Context
var conn   * grpc.ClientConn

type Device struct {
	deviceinfo * importer.DeviceInfo
	eventlist  * importer.EventList
}

var devicemap map[string]* Device

/*///////////////////////////////////////////////////////////////////////*/
// Allows user to register the device for data collection and frequency.
//
//
/*///////////////////////////////////////////////////////////////////////*/
func (s * Device) Attach(aip string, avendor string, afreq uint32) (error, string) {
        fmt.Println("Received Attach\n")
	var default_protocol string	= "https"

	s.deviceinfo = new(importer.DeviceInfo)
	s.eventlist  = new(importer.EventList)
	s.deviceinfo.IpAddress  = aip
	s.deviceinfo.Vendor     = avendor
	s.deviceinfo.Frequency  = afreq
	s.deviceinfo.Protocol   = default_protocol
	_, err := cc.SendDeviceInfo(ctx, s.deviceinfo)

	if err != nil {
		return err ,"attach error!!"
	}else{
		return nil,""
	}
}

/*///////////////////////////////////////////////////////////////////////*/
// Allows user to change the frequency of data collection
//
//
/*///////////////////////////////////////////////////////////////////////*/
func (s * Device) UpdateFreq(wd uint32)(error, string) {
        fmt.Println("Received Period\n")
	s.deviceinfo.Frequency  = wd
	_, err := cc.SetFrequency(ctx, s.deviceinfo)

	if err != nil {
		return err, "period error!!"
	}else{
		return nil,""
	}
}

/*///////////////////////////////////////////////////////////////////////*/
// Allows user to unsubscribe events
//
//
/*///////////////////////////////////////////////////////////////////////*/
func (s * Device) Subscribe(eventlist []string) (error, string) {
        fmt.Println("Received Subscribe\n")
	s.eventlist.Events = eventlist
	s.eventlist.EventIpAddress = s.deviceinfo.IpAddress
	_, err := cc.SubsrcribeGivenEvents(ctx, s.eventlist)

	if err != nil {
		return err, "sub error!!"
	}else{
		return nil,""
	}
}

/*///////////////////////////////////////////////////////////////////////*/
// Allows user to unsubscribe events
//
//
/*///////////////////////////////////////////////////////////////////////*/
func (s * Device) UnSubscribe(eventlist []string) (error, string) {
        fmt.Println("Received UnSubscribe\n")
	s.eventlist.Events = eventlist
	s.eventlist.EventIpAddress = s.deviceinfo.IpAddress
	_, err := cc.UnSubsrcribeGivenEvents(ctx, s.eventlist)

	if err != nil {
		return err, "unsub error!!"
	}else{
		return nil,""
	}
}

/*///////////////////////////////////////////////////////////////////////*/
// Allows user to get the events supported by device
//
//
/*///////////////////////////////////////////////////////////////////////*/
func (s * Device) GetEventSupportList() (error, []string) {
        fmt.Println("Received GetEventSupportList\n")
	var ret_msg * importer.SupportedEventList
	ret_msg, err :=cc.GetEventList(ctx, devicemap[s.deviceinfo.IpAddress].deviceinfo);
	if err != nil {
		return err,ret_msg.Events
	}else{
		fmt.Println("show all event subs:", ret_msg)
		return nil , ret_msg.Events
	}
}

func init() {
        Formatter := new(log.TextFormatter)
        Formatter.TimestampFormat = "02-01-2006 15:04:05"
        Formatter.FullTimestamp = true
        log.SetFormatter(Formatter)
}

func topicListener(topic *string, master sarama.Consumer) {
	log.Info("Starting topicListener for ", *topic)
	consumer, err := master.ConsumePartition(*topic, 0, sarama.OffsetOldest)
	if err != nil {
		log.Error("topicListener panic, topic=[%s]: %s", *topic, err.Error())
		os.Exit(1)
	}
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, os.Interrupt)
	doneCh := make(chan struct{})
	go func() {
		for {
			select {
			case err := <-consumer.Errors():
				log.Error("Consumer error: %s", err.Err)
			case msg := <-consumer.Messages():
				log.Info("Got message on topic=[%s]: %s", *topic, string(msg.Value))
			case <-signals:
				log.Warn("Interrupt is detected")
		                os.Exit(1)
			}
		}
	}()
	<-doneCh
}

func kafkainit() {
	cmd := exec.Command("/bin/sh","kafka_ip.sh")
	var kafkaIP string
	var out bytes.Buffer
	cmd.Stdout = &out
	err := cmd.Run()
	if err != nil {
		log.Info(err)
		os.Exit(1)
	}

	kafkaIP = out.String()
	kafkaIP = strings.TrimSuffix(kafkaIP, "\n")
	kafkaIP = kafkaIP +":9092"
	fmt.Println("IP address of kafka-cord-0:",kafkaIP)
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	master, err := sarama.NewConsumer([]string{kafkaIP}, config)
	if err != nil {
		panic(err)
	}
	DataConsumer = master

	go topicListener(&importerTopic, master)
}
func main() {
	http.DefaultTransport.(*http.Transport).TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	fmt.Println("Launching server...")
	log.Info("kafkaInit starting")
	kafkainit()

	ln, err := net.Listen("tcp", ":9999")
        if err != nil {
                fmt.Println("could not listen")
                log.Fatal("did not listen: %v", err)
        }
	defer ln.Close()

	connS, err := ln.Accept()
	if err != nil {
                fmt.Println("Accept error")
                log.Fatal("Accept error: %v", err)
	}else{

		conn, err = grpc.Dial(default_address, grpc.WithInsecure())
		if err != nil {
		        fmt.Println("could not connect")
			log.Fatal("did not connect: %v", err)
		}
		defer conn.Close()

		cc = importer.NewDeviceManagementClient(conn)
		ctx = context.Background()

		devicemap = make(map[string] *Device)
		loop := true

		for loop == true {
			cmd, _ := bufio.NewReader(connS).ReadString('\n')

			cmd = strings.TrimSuffix(cmd, "\n")
			s := strings.Split(cmd, ":")
			newmessage := "cmd error!!"
			cmd = s[0]

			switch string(cmd) {

			case "attach" :
				cmd_size := len(s)
				var err	error
				var aport   string = default_port
				var avendor string = default_vendor
				var uafreq  uint64 = default_freq

				if (cmd_size == 2 || cmd_size == 5){
					aip    := s[1]
					if(cmd_size == 5){
						aport	= s[2]
						avendor	= s[3]
						afreq	:= s[4]
						uafreq, err = strconv.ParseUint(afreq, 10, 64)

						if err != nil {
							fmt.Print("ParseUint error!!")
						}

						attach_device_ip = aip + ":" + aport
					}else{
						attach_device_ip = aip + ":" + default_port
					}

					if (devicemap[attach_device_ip] == nil){
						dev := new (Device)
						err, newmessage = dev.Attach(attach_device_ip, avendor, uint32(uafreq))
						if err != nil {
							fmt.Print("attach error!!")
						}else{
							fmt.Print("attatch IP:", attach_device_ip)
							newmessage = attach_device_ip
							devicemap[attach_device_ip] = dev
						}
					}else{
						fmt.Print("Change attach IP to %v", attach_device_ip)
						newmessage = attach_device_ip
					}
				}else{
					fmt.Print("Need IP address !!")
					newmessage = "Need IP address !!"

				}
			break

			case "period" :
				cmd_size := len(s)
				if (cmd_size == 2 ){
					if (devicemap[attach_device_ip] != nil){
						pv  := s[1]
						fmt.Print("pv:", pv)
						u, err := strconv.ParseUint(pv, 10, 64)

						if err != nil {
							fmt.Print("ParseUint error!!")
						}else{
							wd := uint32(u)
							dev := devicemap[attach_device_ip]
							err, newmessage =  dev.UpdateFreq(wd)

							if err != nil {
								fmt.Print("period error!!")
							}else{
								newmessage = strings.ToUpper(cmd)
							}
						}
					}else{
						fmt.Print("need attach first!!")
						newmessage = "need attach first!!"
					}
				}else{
					fmt.Print("Need period value !!")
					newmessage = "Need period value !!"
				}

			break

			case "sub","unsub" :
				cmd_size := len(s)
				fmt.Print("cmd is :", cmd)
				if(cmd_size > 4 || cmd_size <0){
					fmt.Print("error event !!")
					newmessage = "error event !!"
				}else{
					var events_list []string
					for i := 1; i < cmd_size; i++ {
						if value, ok := EVENTS_MAP[s[i]]; ok {
							events_list = append(events_list,value)
						} else {
							fmt.Println("key not found")
						}
					}

					if (devicemap[attach_device_ip] != nil){
						dev := devicemap[attach_device_ip]
						if(string(cmd) == "sub"){
							err, newmessage =  dev.Subscribe(events_list)
							if err != nil {
								fmt.Print("sub error!!")
								newmessage = "sub error!!"
							}else{
								newmessage = strings.ToUpper(cmd)
							}
						}else{
							err, newmessage =  dev.UnSubscribe(events_list)
							if err != nil {
								fmt.Print("unsub error!!")
								newmessage = "unsub error!!"
							}else{
								newmessage = strings.ToUpper(cmd)
							}
						}
					}else{
						fmt.Print("need attach first !!")
						newmessage = "need attach first !!"
					}
				}
			break

			case "showeventlist" :
				if (devicemap[attach_device_ip] != nil){
					dev := devicemap[attach_device_ip]
					err, supportlist :=  dev.GetEventSupportList()

					if err != nil {
						fmt.Print("showeventlist error!!")
					}else{
						fmt.Print("showeventlist ", supportlist)
						newmessage = strings.Join(supportlist[:],",")
					}
				}else{
					fmt.Print("need attach first !!")
					newmessage = "need attach first !!"
				}

			break

			case "QUIT" :
				loop = false
	                        newmessage="QUIT"
			break

			default :
			break
			}
			// send string back to client
			connS.Write([]byte(newmessage + "\n"))
	        }
	}
}