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).

    peer.acceptFrame() // HEADERS
    peer.play()
    val connection = connect(peer)

    // Verify the peer received the second ACK.
    val ackFrame = peer.takeFrame()
    assertThat(ackFrame.type).isEqualTo(Http2.TYPE_SETTINGS)
    assertThat(ackFrame.streamId).isEqualTo(0)
    assertThat(ackFrame.ack).isTrue()

    // This stream was created *after* the connection settings were adjusted.

}

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")

	}
	return nil
}

func (x *WorkloadResponse) GetAck() *Ack {
	if x, ok := x.GetPayload().(*WorkloadResponse_Ack); ok {
		return x.Ack
	}
	return nil
}

type isWorkloadResponse_Payload interface {
	isWorkloadResponse_Payload()
}

type WorkloadResponse_Ack struct {
	Ack *Ack `protobuf:"bytes,1,opt,name=ack,proto3,oneof"`
}

func (*WorkloadResponse_Ack) isWorkloadResponse_Payload() {}

      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

				log.Errorf("failed to handle conn: %v", err)
			}
		}()
	}
}

// ZDS protocol is very simple, for every message sent, and ack is sent.
// the ack only has temporal correlation (i.e. it is the first and only ack msg after the message was sent)
// All this to say, that we want to make sure that message to ztunnel are sent from a single goroutine
// so we don't mix messages and acks.

			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{