From b9f94eee1f90070cc4c5779d86fea2693f029356 Mon Sep 17 00:00:00 2001
From: Yuji Oshima <yuji.oshima0x3fd@gmail.com>
Date: Wed, 26 Apr 2017 23:54:27 +0900
Subject: [PATCH] fix

Signed-off-by: Yuji Oshima <yuji.oshima0x3fd@gmail.com>
---
 .../event-repeater/.application_test.go.swp   | Bin 0 -> 20480 bytes
 .../event-repeater/application_test.go        | 204 ++++++++++++++++--
 2 files changed, 182 insertions(+), 22 deletions(-)
 create mode 100644 examples/application/event-repeater/.application_test.go.swp

diff --git a/examples/application/event-repeater/.application_test.go.swp b/examples/application/event-repeater/.application_test.go.swp
new file mode 100644
index 0000000000000000000000000000000000000000..a7ddf0b7db73df1a5b5ae00b16d21d14f2eef0da
GIT binary patch
literal 20480
zcmeI4e~cVu8OMj;tWYGNqL_x4A)Ir)+u7YKML4d+qr1pKuRZRztxen1+1Yn@$K9Qo
z&dgqWxf}+e5G(;Th#`TfC`zKn7!m;!kzfcFO)&mL0wDoSqc$<fFAa#%@AJMhyR+Nf
z-W5^-nN2>O-I@1&=Y5~|*Ymv3(<^1SPEE7V>X#_^oUJIU9#Xcals}$u@FGRAY)*HV
z{WJ~3Up5gE+<*czRVU+nW~OQdwN^<too1%&n04-DEW6?vbt}m5eMZx5a6e<XZo@K-
zz;f&ipXYXv_PEQ90EuIdb_d)K^s2Lx8YF2F34BN;&<ZN)%T_6C$8uScz(wkX?A&v<
zd`N|nG&_+%B7sB#i3Ab}Boas@kVqhrz=u-;fqRnjWi<W7u=#t!>j^#Alf&Qp!mHQw
z{G{;rn(%tK=Xv-uxg-)uB#=lTkw7AWL;{Hf5(y*{NF<O*Adx^KfkXoDfdq`QqI?YR
zeVq(``2QpO|Fw@Q%B$e7;5qOVcpQ8c%mM~pKTA=54(<RYa0Sr8YH$uX4g6=7qWl}Y
z4W0&%f=9qzU>}$UI(YR=lm+er6_5j~!0F%=@QX7P<=fzPa4Wb5WWf30Q{W_U_#=w)
zU2qeqfa}3MDMh&(+z4tQ58gOkQGNg(0Jj4NYzE`t#nTk!1#l4j6dV8(;5_iwsfzL@
zcoF;>d=K0K7+4KX2QQz3dV>4Fz2HW$8C(f+KnEuR1>BEAm0Q48Fa^@!6Ceei$AQYD
z;2U5s*Z@|8S8!1CH}Dtm2>21W6YK+ZupWFGP=6kZ4sn!J?`I+%SWT`^RuD{~t;@1C
zO#Gg<8V$?mrel|fk*VT%%wpr~S@sHCCs<)?VN-F_mTCMN8DWb^k;$+c_c+_jS<RT|
zEU0m2)(qR`4QAM7=5eFEzygOMJ|mTyNc(}|1v9M@ew2BId(3P&KG)n<3Gc1ZJHxN^
z)`@iZV*^UeDXFT%DlOa8hUwMuN0hi})VW5l7Vn=(vo#`QL@!X_2nt@(sPrX}TGFR%
zE6|4ZZ3sfi8-X$1@&j$hF3S!UbE9l*X;|On>=o%#sj2)p!>7^=^f}i<l8UC*96v~3
zTvkV!H9SfsrizYhnd2;2aJjF~7(wl7&uPvDNMF@d%SN+plV?nbBjBpc6Ezf7jn$Nd
z3ORkydM7)$cd{$#9iL)=0umgh4tUd@6L-)F)aA@;*`g!0XaI>`W`5wc%wTb8lm#;$
zuUPx&4rD&ekdVHNGS4-4``nx7o<0{iu4|MUe3Ws|b37VH7!V#0TAs}+R)f2!fPSrG
z*;?81MwyB*qqK!!q%_4CsWghvJ+5I}riK#m+p17e?85cTwi+}nMLxQPW>`)uuo`+1
zt;t)SreeHcEXXJ)QDPcJisr=|95w-k$;$0%SsvF5ds;?A3s3@8MTSEmWJ(UK97;wK
z8tMKHnV|D%mp*6NbumCuVUNr5HaP_n0hK6QLUmq`Qnt8dj8U!0{jj+-LBtdYNB5+?
z5Dn}di#t`{X?Z4xWG!>2I=-{#?K{;`B*FW85{MDsssB!OX_uT9vs@N9&Qy^`rTMkl
zl@(wt7WbuZnaNKU3wVyYqUuBOXVcvGjVecnWHLnvjLF2zak(b~%CO2tqG*e@Vl;e=
z4X2Kl@1`qt;sCjhsk{1=AX`uE?X>OK`=(tXEJh7(Yf%-3|L6TT)~Em8QD5Z?8w)G$
zE1CG{ebq<OV}m7)-s(FG+LG>5+&h?5?c&DX>yFFy={T`)bdOE+<>+#AxZfB!vM8i_
z2UB=5pAY9RX4-%W%xM;G;(NvNHX8<jYMl)OOCl|-eo)2|hHqjP59^xT|Dy~?n>yhV
zfy#7^IjNw;_72hrqDA9~^;4!3t0`5uYuh26r_rif_N3V2x++iG3(8F`(LX6M4%Kny
zisZERH4S%%+{<=}1yPqJVP%^)pq(Li{9RkQ=ObK}m#1uKb``_qw5Zd@up2cB^UjIw
zkd&Vy=_{Jt)|TjPXrI^&r6RM@f?gz6?orkp9;H1FY4Q%;AD6nAcpB4tu=DQN745y~
z5V^~yQZrJ+<K8&a*3j_5=Da&>Q1ob6!wX0`lC38(4)r7zVnN$OMXQDC^;NmV0FCy*
zF>Ec^Ra+%$#WCj1b?0qTrD9zIeNc*dacL2{0lkD$glcN+DO-dtEJRC+0wrBW<0(Q$
zTcD<f`*FY`sp&=1TNLX3NFPTZuM;<>uP+vhg}LHfVRma_7Pr;FbjLFp%L#ODHZ0fY
zy5m*HbC+GVR#g-TLnTd9rK{`{`iky@1u#!vNg>Kani1Ww927=s4gH7(aU%~8%$k=G
z<y00j;Y2|)suYW-4QrBdjhdr3_XHHZ-7xHGPm#TG;cmT}5xtFcVQu9I>MFc%F4;Dz
zHLc(+zH1#*+L-2n(mPXs*@_`)MlKx4*Pf;;$|h1%TVR>ch1;f6#-z%WEZguF=$7d%
zxPg=LV2G&-Rtm=pn5HPMVb-A=VHjJsLiYdXU?-mp`<d+jHMr2{&#>>G05^j=xCUGd
zUV=^k7`Pwo2QGLM_WM7;OW+`Q7<?am9UK5Rg9We=TnR1$|AbBd9C#Sq3%&rRfDTRs
zhftp<zz=~5ieLszgCTG#pgO;Wx+kAR0*M3?2_zCoB#=lTkw7AWL;@?2z!I!5FM{_R
zvu0s2wn_!EoO3XfBD24J!V{TZon%=gLut~*(9SzDBCYD4;=9}lWy`NicS1WG8LX|6
zw2!CqD48tt0VlVq6rJ6|>O>C1UeNRK-;}vF6cr$HM?WZ?ELOCQjMh|2W~4{bn#K|D
zRSPh&G(5<`ow3kxjIu_@k15meIlU-HsVCCSJ@{_ca^qn!qR{M08r|`spahxi{nO}d
z|3(W}M!V5j(dZI<I(61?95<3qkN2jpbB(h4yur<YoCf4)TU4kF+$Y065=oBX2%-B{
zR^F0+v&b8Qo*6hi<iS2n)b`OnQd74d<H5elM&QW9F@NZ38DVM~|9#tS1-yOO9eTL>
z96r9A<3So~0}cn}867pQ_q<Pdd+5Xt52=8oo*;Igo8UB&P9@rBiSX(OUnSk#(6DGp
z(DRm$GZGsesEw_oso$F<dt{P<T@)+0bAZ^g=UOE#nqcs7b!0!H!jbBDFyhH46(ylo
zcD$g}Bt3M4Akm<All^}I7EAIOll>pX_aB0-|1)p^6u}I*4qOXHKy(gp8|;4$z6dS{
zXMi_g-@gE!1%Cj~fFFasU=C~op8;>e&i@nmEqD?<4t@zdunnvO=Yq4qGqCj^0}p}+
zz<zKO_zJib{0rv*&x3D*1uze`fy==q;1uu%eE)w3zXIO@_kr8Mt>6}bKYXBE2R;d2
zMl8TnfZ_x!Fb60;;N##~!~#49?g2Z%Rp0{fHvIpOg1f<f;DOy>JD3HZ2j_z`!7K3p
zKM5WL`@xsM5I7IK4gde|!9(C{papV31s8(V;C%2J;sJgKo(8`G_klaXPOui70S=>I
ze*(S*ZUEZ>_4V1H?+4p~9FEbI#yfVyaF2l_Ha0AAzk67Q3qovI_PBpnI4r~I#Bo68
zd-LtN+&MI*3W;a&nu%?s?IP}~8>T?CSa#1799#+)G7`j^LTOa1Qm1pqG_8Deh&Cr&
z%9u;-@02V&E!D4fyrvOs!Gf*XqipS^C{P~7>9Yue&?uAm#^G2FA8fROpur1v*)ou2
z$VL`g{cYAlDVJU%eZxgmVG760Rq43Qb9-U30atMEfiP(@y;{tXTFNmiGx3c*w~~!L
z(Pk2D=3Q$u$9Z4JoadNqy?gfm&$b?};mO+lzFNDnZaaGiFW$mE2rnHr+|bubYI?_Q
zC~U%_Q|Go`5Wzbb!U#Zuu^?}5GbpN!4Gnc}h3R4eN};2<=~vAf3^;}_c!_XW%2zY7
zMN2&w4pYA>Z$zpj*(<3m`HI7ks}Z$V{im3ns8X~;%5rt+4Zi7FB_2AQ;|&XrXqC7V
z%e^7gwui5P%0VLNw6-gAtf^spPy^8<SOBtBn&So(-9#69L-*6oF8#qNAK`(M<?>Y?
z-sNeOqT^TX#H0yTNJvsd9*TlOLG=kq6OSXoLZ6bp@mL62T)m0n353r*HV!D*0r!O3
z(<lBc$`+|>@#wdehap7KbY$a2mPQnmkfXkjIv<)*+X_^p6k>G5oaCHNQI2!mf{b2B
zb%m3Puu=3gK3&nG81*LI%{yUK!e%>+W-$8(kBT853t;FCN{HeYXl5MOC<+-DA&y>p
z>ktJ-&`6PW7$gxgR=XvVrFGzgED>R243fxlp(0U~NGCd%qBLb-Wjc)5m4PEi5B|WR
m3LQ7d^S?nM#xNG(qhC}jj)QrumJCl_HrlwOP#=eA<NpGNP=Tia

literal 0
HcmV?d00001

diff --git a/examples/application/event-repeater/application_test.go b/examples/application/event-repeater/application_test.go
index 5d1c1660c..1d210f190 100644
--- a/examples/application/event-repeater/application_test.go
+++ b/examples/application/event-repeater/application_test.go
@@ -1,9 +1,12 @@
 package main
 
 import (
+	"crypto/rand"
+	"encoding/binary"
 	"fmt"
 	event_rpc "github.com/docker/infrakit/pkg/rpc/event"
 	rpc_server "github.com/docker/infrakit/pkg/rpc/server"
+	"github.com/docker/infrakit/pkg/spi/application"
 	"github.com/docker/infrakit/pkg/spi/event"
 	testing_event "github.com/docker/infrakit/pkg/testing/event"
 	"github.com/docker/infrakit/pkg/types"
@@ -11,11 +14,71 @@ import (
 	"github.com/stretchr/testify/require"
 	"io/ioutil"
 	"path/filepath"
+	"strconv"
 	"testing"
 	"time"
 )
 
-var MQTTTESTSERVER string = "tcp://test.mosquitto.org:1883"
+var MQTTTESTSERVER = "tcp://iot.eclipse.org:1883"
+var EVENTNUM = 5
+
+func TestUnitUpdate(t *testing.T) {
+	socketPath := tempSocket()
+	stub := func() interface{} { return nil }
+	m := map[string]interface{}{}
+	types.Put(types.PathFromString("test/instance1"), stub, m)
+	types.Put(types.PathFromString("test/instance2"), stub, m)
+	plugin := &testing_event.Plugin{
+		Publisher: (*testing_event.Publisher)(nil),
+		DoList: func(topic types.Path) ([]string, error) {
+			return types.List(topic, m), nil
+		},
+	}
+	var impl rpc_server.VersionedInterface = event_rpc.PluginServerWithTypes(
+		map[string]event.Plugin{
+			"test": plugin,
+		})
+	server, err := rpc_server.StartPluginAtPath(socketPath, impl)
+	require.NoError(t, err)
+	defer server.Stop()
+
+	//Test ADD operation
+	require.NoError(t, err)
+	e := NewEventRepeater(socketPath, "", "stderr", false).(*eventRepeater)
+	mes := &application.Message{
+		Op:       application.ADD,
+		Resource: "event",
+		Data:     types.AnyString("[{\"sourcetopic\":\"test/instance1\",\"sinktopic\":\"test/sink/instance1\"},{\"sourcetopic\":\"test/instance2\",\"sinktopic\":\"test/sink/instance2\"}]"),
+	}
+	err = e.Update(mes)
+	require.NoError(t, err)
+	require.Equal(t, 2, len(e.Events))
+	require.Equal(t, "test/sink/instance1", e.Events["test/instance1"].SinkTopic)
+	require.Equal(t, "test/sink/instance2", e.Events["test/instance2"].SinkTopic)
+
+	//Test DELETE operation
+	mes = &application.Message{
+		Op:       application.DELETE,
+		Resource: "event",
+		Data:     types.AnyString("[{\"sourcetopic\":\"test/instance2\",\"sinktopic\":\"\"}]"),
+	}
+	err = e.Update(mes)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(e.Events))
+	require.Equal(t, "test/sink/instance1", e.Events["test/instance1"].SinkTopic)
+	_, ok := e.Events["test/instance2"]
+	require.Equal(t, false, ok)
+
+	//Test UPDATE operation
+	mes = &application.Message{
+		Op:       application.UPDATE,
+		Resource: "event",
+		Data:     types.AnyString("[{\"sourcetopic\":\"test/instance1\",\"sinktopic\":\"test/event/instance1\"}]"),
+	}
+	err = e.Update(mes)
+	require.NoError(t, err)
+	require.Equal(t, "test/event/instance1", e.Events["test/instance1"].SinkTopic)
+}
 
 func tempSocket() string {
 	dir, err := ioutil.TempDir("", "infrakit-test-")
@@ -24,24 +87,37 @@ func tempSocket() string {
 	}
 	return filepath.Join(dir, "app-impl-test")
 }
-func runEvent(startPub chan struct{}) (*string, rpc_server.Stoppable, error) {
+func runEvent(startPub chan struct{}, tPrefix string) (string, rpc_server.Stoppable, error) {
 	socketPath := tempSocket()
-	events := 5
 	publishChan0 := make(chan chan<- *event.Event)
 	go func() {
 		publish := <-publishChan0
 		defer close(publish)
 		<-startPub
 		// here we have the channel and ready to go
-		for i := 0; i < events; i++ {
+		for i := 0; i < EVENTNUM; i++ {
 			<-time.After(50 * time.Millisecond)
-			fmt.Printf("publish event%d\n", i)
 			publish <- event.Event{
 				Topic: types.PathFromString("instance/create"),
 				ID:    fmt.Sprintf("host-%d", i),
 			}.Init().WithDataMust([]int{1, 2}).Now()
 		}
 	}()
+	publishChan1 := make(chan chan<- *event.Event)
+	go func() {
+		publish := <-publishChan1
+		defer close(publish)
+		<-startPub
+		// here we have the channel and ready to go
+		for i := 0; i < EVENTNUM; i++ {
+			<-time.After(50 * time.Millisecond)
+			publish <- event.Event{
+				Topic: types.PathFromString("instance/create"),
+				ID:    fmt.Sprintf("disk-%d", i),
+			}.Init().WithDataMust([]string{"foo", "bar"}).Now()
+		}
+	}()
+
 	m := map[string]interface{}{}
 	types.Put(types.PathFromString("instance/create"), "instance-create", m)
 	plugin0 := &testing_event.Plugin{
@@ -55,61 +131,145 @@ func runEvent(startPub chan struct{}) (*string, rpc_server.Stoppable, error) {
 			},
 		},
 	}
+	plugin1 := &testing_event.Plugin{
+		DoList: func(topic types.Path) ([]string, error) {
+			return types.List(topic, m), nil
+		},
+		Publisher: &testing_event.Publisher{
+			DoPublishOn: func(c chan<- *event.Event) {
+				publishChan1 <- c
+				close(publishChan1)
+			},
+		},
+	}
 	var impl rpc_server.VersionedInterface = event_rpc.PluginServerWithTypes(
 		map[string]event.Plugin{
-			"iktest": plugin0,
+			tPrefix + "-compute": plugin0,
+			tPrefix + "-storage": plugin1,
 		})
 	server, err := rpc_server.StartPluginAtPath(socketPath, impl)
 	if err != nil {
-		return nil, nil, err
+		return "", nil, err
 	}
-	return &socketPath, server, nil
+	return socketPath, server, nil
 }
 
-func runSub(msgch chan MQTT.Message) (MQTT.Client, error) {
+func runSub(msgch chan MQTT.Message, tPrefix string) (MQTT.Client, error) {
 	opts := MQTT.NewClientOptions().AddBroker(MQTTTESTSERVER)
 	client := MQTT.NewClient(opts)
 	if token := client.Connect(); token.Wait() && token.Error() != nil {
 		return nil, token.Error()
 	}
 	subToken := client.Subscribe(
-		"iktest/instance/create",
+		tPrefix+"/instance/create",
 		0,
 		func(client MQTT.Client, msg MQTT.Message) {
 			msgch <- msg
 		})
-	fmt.Printf("mqtt substart")
 	if subToken.Wait() && subToken.Error() != nil {
 		return nil, subToken.Error()
 	}
 	return client, nil
 }
 
-func TestIntegration(t *testing.T) {
+func TestIntegrationAllowAll(t *testing.T) {
+	var n uint64
+	binary.Read(rand.Reader, binary.LittleEndian, &n)
+	randString := strconv.FormatUint(n, 36)
+	topicPrefix := "ifktest-" + randString
 	startPub := make(chan struct{})
-	socketPath, erpcsrv, err := runEvent(startPub)
+	socketPath, erpcsrv, err := runEvent(startPub, topicPrefix)
 	defer erpcsrv.Stop()
 	require.NoError(t, err)
-	mqsubch := make(chan MQTT.Message)
-	mqttClient, err := runSub(mqsubch)
+	mqsubch0 := make(chan MQTT.Message)
+	mqttClient0, err := runSub(mqsubch0, topicPrefix+"-compute")
+	require.NoError(t, err)
+	mqsubch1 := make(chan MQTT.Message)
+	mqttClient1, err := runSub(mqsubch1, topicPrefix+"-storage")
 	require.NoError(t, err)
-	defer mqttClient.Disconnect(250)
-	app := NewEventRepeater(*socketPath, MQTTTESTSERVER, "mqtt", true)
+	defer mqttClient0.Disconnect(250)
+	defer mqttClient1.Disconnect(250)
+	app := NewEventRepeater(socketPath, MQTTTESTSERVER, "mqtt", true)
 	defer app.(*eventRepeater).Stop()
 	close(startPub)
-	var subEvent int = 0
+	var subEvent0 int
+	var subEvent1 int
 loop:
 	for {
 		select {
 		case <-time.After(500 * time.Millisecond):
 			break loop
-		case sub := <-mqsubch:
-			subany := types.AnyBytes(sub.Payload())
+		case sub0 := <-mqsubch0:
+			subany := types.AnyBytes(sub0.Payload())
+			subevent := event.Event{}
+			err := subany.Decode(&subevent)
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("host-%d", subEvent0), subevent.ID)
+			subEvent0++
+		case sub1 := <-mqsubch1:
+			subany := types.AnyBytes(sub1.Payload())
+			subevent := event.Event{}
+			err := subany.Decode(&subevent)
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("disk-%d", subEvent1), subevent.ID)
+			subEvent1++
+
+		}
+	}
+	require.Equal(t, EVENTNUM, subEvent0)
+	require.Equal(t, EVENTNUM, subEvent1)
+}
+
+func TestIntegrationDenyAll(t *testing.T) {
+	var n uint64
+	binary.Read(rand.Reader, binary.LittleEndian, &n)
+	randString := strconv.FormatUint(n, 36)
+	topicPrefix := "ifktest-" + randString
+	startPub := make(chan struct{})
+	socketPath, erpcsrv, err := runEvent(startPub, topicPrefix)
+	defer erpcsrv.Stop()
+	require.NoError(t, err)
+	mqsubch0 := make(chan MQTT.Message)
+	mqttClient0, err := runSub(mqsubch0, topicPrefix+"-compute")
+	require.NoError(t, err)
+	mqsubch1 := make(chan MQTT.Message)
+	mqttClient1, err := runSub(mqsubch1, topicPrefix+"-storage")
+	require.NoError(t, err)
+	defer mqttClient0.Disconnect(250)
+	defer mqttClient1.Disconnect(250)
+	app := NewEventRepeater(socketPath, MQTTTESTSERVER, "mqtt", false)
+	defer app.(*eventRepeater).Stop()
+	m := &application.Message{
+		Op:       application.ADD,
+		Resource: "event",
+		Data:     types.AnyString("[{\"sourcetopic\":\"" + topicPrefix + "-compute/instance/create\",\"sinktopic\":\"\"}]"),
+	}
+	err = app.Update(m)
+	require.NoError(t, err)
+	close(startPub)
+	var subEvent0 int
+	var subEvent1 int
+loop:
+	for {
+		select {
+		case <-time.After(500 * time.Millisecond):
+			break loop
+		case sub0 := <-mqsubch0:
+			subany := types.AnyBytes(sub0.Payload())
+			subevent := event.Event{}
+			err := subany.Decode(&subevent)
+			require.NoError(t, err)
+			require.Equal(t, fmt.Sprintf("host-%d", subEvent0), subevent.ID)
+			subEvent0++
+		case sub1 := <-mqsubch1:
+			subany := types.AnyBytes(sub1.Payload())
 			subevent := event.Event{}
 			err := subany.Decode(&subevent)
 			require.NoError(t, err)
-			require.Equal(t, subevent.ID, fmt.Sprintf("host-%d", subEvent))
-			subEvent++
+			require.Equal(t, fmt.Sprintf("disk-%d", subEvent1), subevent.ID)
+			subEvent1++
 		}
 	}
+	require.Equal(t, EVENTNUM, subEvent0)
+	require.Equal(t, 0, subEvent1)
 }