John Howard <******@****.***> 1656720476 +0000

message SnapshotSent {
}

// Ztunnel ack message. If error is not empty, this is an error message.
message Ack {
  string error = 1;
}

/*
Protocol details:
on new connection from ztunnel to CNI, the CNI agent
- will send all the existing payloads (that it has in its cache) to the ztunnel using AddWorkload message.
- the ztunnel will send an ack for each payload (which the CNI will wait for before sending the next one).

}

func sendAck(c *net.UnixConn) {
	ack := &zdsapi.WorkloadResponse{
		Payload: &zdsapi.WorkloadResponse_Ack{
			Ack: &zdsapi.Ack{},
		},
	}
	data, err := proto.Marshal(ack)
	if err != nil {
		panic(err)
	}
	err = c.SetWriteDeadline(time.Now().Add(time.Second))
	if err != nil {
		panic(err)
	}
	c.Write(data)
}

func sendHello(c *net.UnixConn) {
	ack := &zdsapi.ZdsHello{

      .isEqualTo("<< 0x00000000    15 SETTINGS      ")
    assertThat(frameLog(false, 0, 0, TYPE_SETTINGS, FLAG_ACK))
      .isEqualTo(">> 0x00000000     0 SETTINGS      ACK")
    assertThat(frameLog(true, 0, 0, TYPE_SETTINGS, FLAG_ACK))
      .isEqualTo("<< 0x00000000     0 SETTINGS      ACK")
    assertThat(frameLog(true, 3, 22, TYPE_HEADERS, FLAG_END_HEADERS))
      .isEqualTo("<< 0x00000003    22 HEADERS       END_HEADERS")

      this.streamId = streamId
      this.errorCode = errorCode
    }

    override fun ping(
      ack: Boolean,
      payload1: Int,
      payload2: Int,
    ) {
      check(type == -1)
      type = Http2.TYPE_PING
      this.ack = ack
      this.payload1 = payload1
      this.payload2 = payload2
    }

    override fun goAway(
      lastGoodStreamId: Int,

                    project.extensions.create("test", TestExtension::class)
                }
            }

            open class TestExtension {
                fun ack() = println("Ack!")
            }
            """
        )

        withFile(
            "src/test/kotlin/plugin/TestPluginTest.kt",
            """

            package plugin

			response: true,
		},
		{
			name: "ack",
			proxy: &TestProxy{
				WatchedResources: map[string]*WatchedResource{
					model.ClusterType: {
						NonceSent: "nonce",
					},
				},
			},
			request: &discovery.DiscoveryRequest{
				TypeUrl:       model.ClusterType,
				VersionInfo:   "v1",
				ResponseNonce: "nonce",
			},
			response: false,
		},
		{
			name: "ack forced",
			proxy: &TestProxy{

[2020-01-01 00:00:00] >> 0x00000003    47 HEADERS       END_STREAM|END_HEADERS
[2020-01-01 00:00:00] << 0x00000000     6 SETTINGS
[2020-01-01 00:00:00] << 0x00000000     0 SETTINGS      ACK
[2020-01-01 00:00:00] << 0x00000000     4 WINDOW_UPDATE
[2020-01-01 00:00:00] >> 0x00000000     0 SETTINGS      ACK
[2020-01-01 00:00:00] << 0x00000003   322 HEADERS       END_HEADERS
[2020-01-01 00:00:00] << 0x00000003   288 DATA

	}
	// Acknowledge Mux created.
	// Send async.
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		var ack message
		ack.Op = OpAckMux
		ack.Flags = m.BaseFlags
		ack.MuxID = m.ID
		m.send(ack)
		if debugPrint {
			fmt.Println("connected stream mux:", ack.MuxID)
		}
	}()

	// Data inbound to the handler
	var handlerIn chan []byte
	if inboundCap > 0 {

	if ok {
		return Key[O](keyFunc(ac.Name, ac.Namespace)), true
	}
	arn, ok := any(a).(ResourceNamer)
	if ok {
		return Key[O](arn.ResourceName()), true
	}
	ack := GetApplyConfigKey(a)
	if ack != nil {
		return *ack, true
	}
	return "", false
}

// getLabels returns the labels for an object, if possible.
// Warning: this will panic if the labels is not available.