struct wreplsrv_out_connect_state {
enum wreplsrv_out_connect_stage stage;
struct composite_context *c;
- struct wrepl_request *req;
- struct composite_context *c_req;
struct wrepl_associate assoc_io;
enum winsrepl_partner_type type;
struct wreplsrv_out_connection *wreplconn;
+ struct tevent_req *subreq;
};
-static void wreplsrv_out_connect_handler_creq(struct composite_context *c_req);
-static void wreplsrv_out_connect_handler_req(struct wrepl_request *req);
+static void wreplsrv_out_connect_handler_treq(struct tevent_req *subreq);
static NTSTATUS wreplsrv_out_connect_wait_socket(struct wreplsrv_out_connect_state *state)
{
NTSTATUS status;
- status = wrepl_connect_recv(state->c_req);
+ status = wrepl_connect_recv(state->subreq);
+ TALLOC_FREE(state->subreq);
NT_STATUS_NOT_OK_RETURN(status);
- state->req = wrepl_associate_send(state->wreplconn->sock, &state->assoc_io);
- NT_STATUS_HAVE_NO_MEMORY(state->req);
+ state->subreq = wrepl_associate_send(state,
+ state->wreplconn->service->task->event_ctx,
+ state->wreplconn->sock, &state->assoc_io);
+ NT_STATUS_HAVE_NO_MEMORY(state->subreq);
- state->req->async.fn = wreplsrv_out_connect_handler_req;
- state->req->async.private_data = state;
+ tevent_req_set_callback(state->subreq,
+ wreplsrv_out_connect_handler_treq,
+ state);
state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_ASSOC_CTX;
{
NTSTATUS status;
- status = wrepl_associate_recv(state->req, &state->assoc_io);
+ status = wrepl_associate_recv(state->subreq, &state->assoc_io);
+ TALLOC_FREE(state->subreq);
NT_STATUS_NOT_OK_RETURN(status);
state->wreplconn->assoc_ctx.peer_ctx = state->assoc_io.out.assoc_ctx;
}
}
-static void wreplsrv_out_connect_handler_creq(struct composite_context *creq)
-{
- struct wreplsrv_out_connect_state *state = talloc_get_type(creq->async.private_data,
- struct wreplsrv_out_connect_state);
- wreplsrv_out_connect_handler(state);
- return;
-}
-
-static void wreplsrv_out_connect_handler_req(struct wrepl_request *req)
+static void wreplsrv_out_connect_handler_treq(struct tevent_req *subreq)
{
- struct wreplsrv_out_connect_state *state = talloc_get_type(req->async.private_data,
+ struct wreplsrv_out_connect_state *state = tevent_req_callback_data(subreq,
struct wreplsrv_out_connect_state);
wreplsrv_out_connect_handler(state);
return;
/* we have a connection already, so use it */
if (wreplconn) {
- if (!wreplconn->sock->dead) {
+ if (wrepl_socket_is_connected(wreplconn->sock)) {
state->stage = WREPLSRV_OUT_CONNECT_STAGE_DONE;
state->wreplconn= wreplconn;
composite_done(c);
wreplconn->service = service;
wreplconn->partner = partner;
- wreplconn->sock = wrepl_socket_init(wreplconn, service->task->event_ctx, lp_iconv_convenience(service->task->lp_ctx));
+ wreplconn->sock = wrepl_socket_init(wreplconn, service->task->event_ctx);
if (!wreplconn->sock) goto failed;
state->stage = WREPLSRV_OUT_CONNECT_STAGE_WAIT_SOCKET;
state->wreplconn= wreplconn;
- state->c_req = wrepl_connect_send(wreplconn->sock,
+ state->subreq = wrepl_connect_send(state,
+ service->task->event_ctx,
+ wreplconn->sock,
partner->our_address?partner->our_address:wrepl_best_ip(service->task->lp_ctx, partner->address),
partner->address);
- if (!state->c_req) goto failed;
+ if (!state->subreq) goto failed;
- state->c_req->async.fn = wreplsrv_out_connect_handler_creq;
- state->c_req->async.private_data = state;
+ tevent_req_set_callback(state->subreq,
+ wreplsrv_out_connect_handler_treq,
+ state);
return c;
failed:
*wreplconn = talloc_reference(mem_ctx, state->wreplconn);
if (!*wreplconn) status = NT_STATUS_NO_MEMORY;
} else {
- status = NT_STATUS_INVALID_CONNECTION;
+ status = NT_STATUS_CONNECTION_DISCONNECTED;
}
}
struct wreplsrv_pull_table_state {
enum wreplsrv_pull_table_stage stage;
struct composite_context *c;
- struct wrepl_request *req;
struct wrepl_pull_table table_io;
struct wreplsrv_pull_table_io *io;
struct composite_context *creq;
struct wreplsrv_out_connection *wreplconn;
+ struct tevent_req *subreq;
};
-static void wreplsrv_pull_table_handler_req(struct wrepl_request *req);
+static void wreplsrv_pull_table_handler_treq(struct tevent_req *subreq);
static NTSTATUS wreplsrv_pull_table_wait_connection(struct wreplsrv_pull_table_state *state)
{
NT_STATUS_NOT_OK_RETURN(status);
state->table_io.in.assoc_ctx = state->wreplconn->assoc_ctx.peer_ctx;
- state->req = wrepl_pull_table_send(state->wreplconn->sock, &state->table_io);
- NT_STATUS_HAVE_NO_MEMORY(state->req);
+ state->subreq = wrepl_pull_table_send(state,
+ state->wreplconn->service->task->event_ctx,
+ state->wreplconn->sock, &state->table_io);
+ NT_STATUS_HAVE_NO_MEMORY(state->subreq);
- state->req->async.fn = wreplsrv_pull_table_handler_req;
- state->req->async.private_data = state;
+ tevent_req_set_callback(state->subreq,
+ wreplsrv_pull_table_handler_treq,
+ state);
state->stage = WREPLSRV_PULL_TABLE_STAGE_WAIT_TABLE_REPLY;
{
NTSTATUS status;
- status = wrepl_pull_table_recv(state->req, state, &state->table_io);
+ status = wrepl_pull_table_recv(state->subreq, state, &state->table_io);
+ TALLOC_FREE(state->subreq);
NT_STATUS_NOT_OK_RETURN(status);
state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
return;
}
-static void wreplsrv_pull_table_handler_req(struct wrepl_request *req)
+static void wreplsrv_pull_table_handler_treq(struct tevent_req *subreq)
{
- struct wreplsrv_pull_table_state *state = talloc_get_type(req->async.private_data,
+ struct wreplsrv_pull_table_state *state = tevent_req_callback_data(subreq,
struct wreplsrv_pull_table_state);
wreplsrv_pull_table_handler(state);
return;
c->private_data = state;
if (io->in.num_owners) {
+ struct wrepl_wins_owner *partners;
+ uint32_t i;
+
+ partners = talloc_array(state,
+ struct wrepl_wins_owner,
+ io->in.num_owners);
+ if (composite_nomem(partners, c)) goto failed;
+
+ for (i=0; i < io->in.num_owners; i++) {
+ partners[i] = io->in.owners[i];
+ partners[i].address = talloc_strdup(partners,
+ io->in.owners[i].address);
+ if (composite_nomem(partners[i].address, c)) goto failed;
+ }
+
state->table_io.out.num_partners = io->in.num_owners;
- state->table_io.out.partners = io->in.owners;
+ state->table_io.out.partners = partners;
state->stage = WREPLSRV_PULL_TABLE_STAGE_DONE;
composite_done(c);
return c;
struct wreplsrv_pull_table_state *state = talloc_get_type(c->private_data,
struct wreplsrv_pull_table_state);
io->out.num_owners = state->table_io.out.num_partners;
- io->out.owners = talloc_reference(mem_ctx, state->table_io.out.partners);
+ io->out.owners = talloc_move(mem_ctx, &state->table_io.out.partners);
}
talloc_free(c);
struct wreplsrv_pull_names_state *state = talloc_get_type(c->private_data,
struct wreplsrv_pull_names_state);
io->out.num_names = state->pull_io.out.num_names;
- io->out.names = talloc_reference(mem_ctx, state->pull_io.out.names);
+ io->out.names = talloc_move(mem_ctx, &state->pull_io.out.names);
}
talloc_free(c);
struct wreplsrv_pull_names_io names_io;
struct composite_context *creq;
struct wrepl_associate_stop assoc_stop_io;
- struct wrepl_request *req;
+ struct tevent_req *subreq;
};
static void wreplsrv_pull_cycle_handler_creq(struct composite_context *creq);
-static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req);
+static void wreplsrv_pull_cycle_handler_treq(struct tevent_req *subreq);
static NTSTATUS wreplsrv_pull_cycle_next_owner_do_work(struct wreplsrv_pull_cycle_state *state)
{
if (state->stage == WREPLSRV_PULL_CYCLE_STAGE_DONE && state->io->in.wreplconn) {
state->assoc_stop_io.in.assoc_ctx = state->io->in.wreplconn->assoc_ctx.peer_ctx;
state->assoc_stop_io.in.reason = 0;
- state->req = wrepl_associate_stop_send(state->io->in.wreplconn->sock, &state->assoc_stop_io);
- NT_STATUS_HAVE_NO_MEMORY(state->req);
+ state->subreq = wrepl_associate_stop_send(state,
+ state->io->in.wreplconn->service->task->event_ctx,
+ state->io->in.wreplconn->sock,
+ &state->assoc_stop_io);
+ NT_STATUS_HAVE_NO_MEMORY(state->subreq);
- state->req->async.fn = wreplsrv_pull_cycle_handler_req;
- state->req->async.private_data = state;
+ tevent_req_set_callback(state->subreq,
+ wreplsrv_pull_cycle_handler_treq,
+ state);
state->stage = WREPLSRV_PULL_CYCLE_STAGE_WAIT_STOP_ASSOC;
}
{
NTSTATUS status;
- status = wrepl_associate_stop_recv(state->req, &state->assoc_stop_io);
+ status = wrepl_associate_stop_recv(state->subreq, &state->assoc_stop_io);
+ TALLOC_FREE(state->subreq);
NT_STATUS_NOT_OK_RETURN(status);
state->stage = WREPLSRV_PULL_CYCLE_STAGE_DONE;
return;
}
-static void wreplsrv_pull_cycle_handler_req(struct wrepl_request *req)
+static void wreplsrv_pull_cycle_handler_treq(struct tevent_req *subreq)
{
- struct wreplsrv_pull_cycle_state *state = talloc_get_type(req->async.private_data,
+ struct wreplsrv_pull_cycle_state *state = tevent_req_callback_data(subreq,
struct wreplsrv_pull_cycle_state);
wreplsrv_pull_cycle_handler(state);
return;
enum wreplsrv_push_notify_stage {
WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT,
+ WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE,
WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM,
WREPLSRV_PUSH_NOTIFY_STAGE_DONE
};
enum wrepl_replication_cmd command;
bool full_table;
struct wrepl_send_ctrl ctrl;
- struct wrepl_request *req;
struct wrepl_packet req_packet;
struct wrepl_packet *rep_packet;
struct composite_context *creq;
struct wreplsrv_out_connection *wreplconn;
+ struct tevent_req *subreq;
};
static void wreplsrv_push_notify_handler_creq(struct composite_context *creq);
-static void wreplsrv_push_notify_handler_req(struct wrepl_request *req);
+static void wreplsrv_push_notify_handler_treq(struct tevent_req *subreq);
static NTSTATUS wreplsrv_push_notify_update(struct wreplsrv_push_notify_state *state)
{
struct wrepl_packet *req = &state->req_packet;
struct wrepl_replication *repl_out = &state->req_packet.message.replication;
struct wrepl_table *table_out = &state->req_packet.message.replication.info.table;
- struct wreplsrv_in_connection *wrepl_in;
NTSTATUS status;
- struct socket_context *sock;
/* prepare the outgoing request */
req->opcode = WREPL_OPCODE_BITS;
NT_STATUS_NOT_OK_RETURN(status);
/* queue the request */
- state->req = wrepl_request_send(state->wreplconn->sock, req, NULL);
- NT_STATUS_HAVE_NO_MEMORY(state->req);
-
- /*
- * now we need to convert the wrepl_socket (client connection)
- * into a wreplsrv_in_connection (server connection), because
- * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
- * message is received by the peer.
- */
-
- /* steal the socket_context */
- sock = state->wreplconn->sock->sock;
- state->wreplconn->sock->sock = NULL;
- talloc_steal(state, sock);
-
- /*
- * TODO: steal the tstream if we switch the client to tsocket.
- * This is just to get a compiler error as soon as we remove
- * packet_context.
- */
- state->wreplconn->sock->packet = NULL;
-
- /*
- * free the wrepl_socket (client connection)
- */
- talloc_free(state->wreplconn->sock);
- state->wreplconn->sock = NULL;
-
- /*
- * now create a wreplsrv_in_connection,
- * on which we act as server
- *
- * NOTE: sock and packet will be stolen by
- * wreplsrv_in_connection_merge()
- */
- status = wreplsrv_in_connection_merge(state->io->in.partner,
- sock, &wrepl_in);
- NT_STATUS_NOT_OK_RETURN(status);
-
- wrepl_in->assoc_ctx.peer_ctx = state->wreplconn->assoc_ctx.peer_ctx;
- wrepl_in->assoc_ctx.our_ctx = 0;
+ state->subreq = wrepl_request_send(state,
+ state->wreplconn->service->task->event_ctx,
+ state->wreplconn->sock, req, NULL);
+ NT_STATUS_HAVE_NO_MEMORY(state->subreq);
- /* now we can free the wreplsrv_out_connection */
- TALLOC_FREE(state->wreplconn);
+ tevent_req_set_callback(state->subreq,
+ wreplsrv_push_notify_handler_treq,
+ state);
- state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
+ state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE;
return NT_STATUS_OK;
}
/* we won't get a reply to a inform message */
state->ctrl.send_only = true;
- state->req = wrepl_request_send(state->wreplconn->sock, req, &state->ctrl);
- NT_STATUS_HAVE_NO_MEMORY(state->req);
+ state->subreq = wrepl_request_send(state,
+ state->wreplconn->service->task->event_ctx,
+ state->wreplconn->sock, req, &state->ctrl);
+ NT_STATUS_HAVE_NO_MEMORY(state->subreq);
- state->req->async.fn = wreplsrv_push_notify_handler_req;
- state->req->async.private_data = state;
+ tevent_req_set_callback(state->subreq,
+ wreplsrv_push_notify_handler_treq,
+ state);
state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM;
default:
return NT_STATUS_INTERNAL_ERROR;
}
+}
+
+static NTSTATUS wreplsrv_push_notify_wait_update(struct wreplsrv_push_notify_state *state)
+{
+ struct wreplsrv_in_connection *wrepl_in;
+ struct tstream_context *stream;
+ void *process_context = NULL;
+ NTSTATUS status;
+
+ status = wrepl_request_recv(state->subreq, state, NULL);
+ TALLOC_FREE(state->subreq);
+ NT_STATUS_NOT_OK_RETURN(status);
+
+ /*
+ * now we need to convert the wrepl_socket (client connection)
+ * into a wreplsrv_in_connection (server connection), because
+ * we'll act as a server on this connection after the WREPL_REPL_UPDATE*
+ * message is received by the peer.
+ */
+
+ status = wrepl_socket_split_stream(state->wreplconn->sock, state, &stream);
+ NT_STATUS_NOT_OK_RETURN(status);
- return NT_STATUS_INTERNAL_ERROR;
+ /*
+ * now create a wreplsrv_in_connection,
+ * on which we act as server
+ *
+ * NOTE: stream will be stolen by
+ * wreplsrv_in_connection_merge()
+ */
+ process_context = state->io->in.partner->service->task->process_context;
+ status = wreplsrv_in_connection_merge(state->io->in.partner,
+ state->wreplconn->assoc_ctx.peer_ctx,
+ &stream,
+ &wrepl_in, process_context);
+ NT_STATUS_NOT_OK_RETURN(status);
+
+ /* now we can free the wreplsrv_out_connection */
+ TALLOC_FREE(state->wreplconn);
+
+ state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
+ return NT_STATUS_OK;
}
static NTSTATUS wreplsrv_push_notify_wait_inform(struct wreplsrv_push_notify_state *state)
{
NTSTATUS status;
- status = wrepl_request_recv(state->req, state, NULL);
+ status = wrepl_request_recv(state->subreq, state, NULL);
+ TALLOC_FREE(state->subreq);
NT_STATUS_NOT_OK_RETURN(status);
state->stage = WREPLSRV_PUSH_NOTIFY_STAGE_DONE;
case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_CONNECT:
c->status = wreplsrv_push_notify_wait_connect(state);
break;
+ case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_UPDATE:
+ c->status = wreplsrv_push_notify_wait_update(state);
+ break;
case WREPLSRV_PUSH_NOTIFY_STAGE_WAIT_INFORM:
c->status = wreplsrv_push_notify_wait_inform(state);
break;
return;
}
-static void wreplsrv_push_notify_handler_req(struct wrepl_request *req)
+static void wreplsrv_push_notify_handler_treq(struct tevent_req *subreq)
{
- struct wreplsrv_push_notify_state *state = talloc_get_type(req->async.private_data,
+ struct wreplsrv_push_notify_state *state = tevent_req_callback_data(subreq,
struct wreplsrv_push_notify_state);
wreplsrv_push_notify_handler(state);
return;
git.samba.org / vlendec / samba-autobuild / .git / blobdiff